From 66a6674a6a7708753edd30bbb63a10034d88bba9 Mon Sep 17 00:00:00 2001 From: ycc Date: Sat, 28 Feb 2026 10:08:55 +0100 Subject: [PATCH] ProcessSentMessages added for server acks --- client/helpers/bgSendHelper.go | 2 +- client/helpers/messageHelper.go | 77 +++++++++++++++++++++++++++++++++ client/messagestorage.go | 11 +++++ client/sendjobs.go | 68 +++++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 1 deletion(-) diff --git a/client/helpers/bgSendHelper.go b/client/helpers/bgSendHelper.go index 0ca62b4..7aa0c6e 100644 --- a/client/helpers/bgSendHelper.go +++ b/client/helpers/bgSendHelper.go @@ -88,7 +88,7 @@ func processSendQueue(storagePath, queue string) int { serverIdx, sendErr := attemptSendJob(job) if sendErr == nil { - now := time.Now() + now := time.Now().UTC() job.Status = client.SendStatusSent job.SentAt = &now job.SuccessfulServer = &serverIdx diff --git a/client/helpers/messageHelper.go b/client/helpers/messageHelper.go index 7660e7f..574f32d 100644 --- a/client/helpers/messageHelper.go +++ b/client/helpers/messageHelper.go @@ -1,6 +1,11 @@ package helpers import ( + "os" + "path/filepath" + "strconv" + "strings" + "forge.redroom.link/yves/meowlib" "forge.redroom.link/yves/meowlib/client" ) @@ -93,3 +98,75 @@ func BuildAckMessage(messageUid string, srvuid string, peer_uid string, received func ReadAckMessageResponse() { //! update the status in message store } + +// ProcessSentMessages scans every send queue under storagePath/queues/, updates +// the message storage entry with server delivery info for each sent job, then +// removes the job from the queue. Returns the number of messages updated. +// +// Callers must follow two conventions when building a SendJob: +// - job.Queue = peer UID (used to look up the peer and its DB files) +// - job.File = a path whose basename without extension is the message's +// SQLite row ID as a decimal integer (e.g. "42.bin") +func ProcessSentMessages(storagePath, password string) int { + queueDir := filepath.Join(storagePath, "queues") + entries, err := os.ReadDir(queueDir) + if err != nil { + logger.Warn().Err(err).Str("dir", queueDir).Msg("ProcessSentMessages: ReadDir") + return 0 + } + + updated := 0 + identity := client.GetConfig().GetIdentity() + + for _, entry := range entries { + if entry.IsDir() { + continue + } + queue := entry.Name() + + jobs, err := client.GetSentJobs(storagePath, queue) + if err != nil { + logger.Error().Err(err).Str("queue", queue).Msg("ProcessSentMessages: GetSentJobs") + continue + } + + for _, job := range jobs { + if job.SuccessfulServer == nil || job.SentAt == nil { + // No delivery info – discard the job so it doesn't block the queue + if err := client.DeleteSendJob(storagePath, queue, job.ID); err != nil { + logger.Error().Err(err).Int64("id", job.ID).Msg("ProcessSentMessages: DeleteSendJob (incomplete)") + } + continue + } + + // Resolve the peer from the queue name to get its DB file list + peer := identity.Peers.GetFromUid(queue) + if peer == nil || len(peer.DbIds) == 0 { + logger.Warn().Str("queue", queue).Msg("ProcessSentMessages: peer not found or has no DB") + continue + } + dbFile := peer.DbIds[len(peer.DbIds)-1] + + // Parse the DB row ID from the job file's basename (e.g. "42.bin" → 42) + base := strings.TrimSuffix(filepath.Base(job.File), filepath.Ext(job.File)) + dbId, err := strconv.ParseInt(base, 10, 64) + if err != nil { + logger.Error().Err(err).Str("file", job.File).Msg("ProcessSentMessages: parse dbId from filename") + continue + } + + serverUid := job.Servers[*job.SuccessfulServer].GetUid() + receiveTime := uint64(job.SentAt.Unix()) + + if err := client.SetMessageServerDelivery(dbFile, dbId, serverUid, receiveTime, password); err != nil { + logger.Error().Err(err).Str("queue", queue).Int64("dbId", dbId).Msg("ProcessSentMessages: SetMessageServerDelivery") + continue + } + if err := client.DeleteSendJob(storagePath, queue, job.ID); err != nil { + logger.Error().Err(err).Int64("id", job.ID).Msg("ProcessSentMessages: DeleteSendJob") + } + updated++ + } + } + return updated +} diff --git a/client/messagestorage.go b/client/messagestorage.go index ce3d801..2b123d0 100644 --- a/client/messagestorage.go +++ b/client/messagestorage.go @@ -353,6 +353,17 @@ func getMessageCount(dbid string) (int, error) { return count, nil } +// SetMessageServerDelivery updates the server delivery UUID and timestamp for an existing stored message. +func SetMessageServerDelivery(dbFile string, dbId int64, serverUid string, receiveTime uint64, password string) error { + dbm, err := GetDbMessage(dbFile, dbId, password) + if err != nil { + return err + } + dbm.ServerDeliveryUuid = serverUid + dbm.ServerDeliveryTimestamp = receiveTime + return UpdateDbMessage(dbm, dbFile, dbId, password) +} + func createMessageTable(db *sql.DB) error { createMessageTableSQL := `CREATE TABLE message ( "id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, diff --git a/client/sendjobs.go b/client/sendjobs.go index cc1ac3b..7e73061 100644 --- a/client/sendjobs.go +++ b/client/sendjobs.go @@ -258,6 +258,74 @@ func GetSendJob(storagePath, queue string, id int64) (*SendJob, error) { return job, nil } +// GetSentJobs returns all successfully-sent jobs from the named queue, +// ordered oldest first. Use this to reconcile delivery status with the +// message store and clean up completed entries. +func GetSentJobs(storagePath, queue string) ([]*SendJob, error) { + db, err := openOrCreateSendQueue(sendQueueDbPath(storagePath, queue)) + if err != nil { + return nil, err + } + defer db.Close() + + rows, err := db.Query( + `SELECT id, file, servers, timeout, inserted_at, sent_at, retries, successful_server + FROM queue WHERE status = ? ORDER BY id ASC`, + SendStatusSent, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var jobs []*SendJob + for rows.Next() { + var ( + id int64 + file string + serversJSON string + timeout int + insertedAt int64 + sentAt sql.NullInt64 + retriesJSON string + successfulServer sql.NullInt64 + ) + if err := rows.Scan(&id, &file, &serversJSON, &timeout, &insertedAt, &sentAt, &retriesJSON, &successfulServer); err != nil { + return nil, err + } + + var servers []Server + if err := json.Unmarshal([]byte(serversJSON), &servers); err != nil { + return nil, err + } + var retries []int + if err := json.Unmarshal([]byte(retriesJSON), &retries); err != nil { + return nil, err + } + + job := &SendJob{ + ID: id, + Queue: queue, + File: file, + Servers: servers, + Timeout: timeout, + InsertedAt: time.Unix(insertedAt, 0), + Status: SendStatusSent, + Retries: retries, + } + if sentAt.Valid { + t := time.Unix(sentAt.Int64, 0) + job.SentAt = &t + } + if successfulServer.Valid { + v := int(successfulServer.Int64) + job.SuccessfulServer = &v + } + jobs = append(jobs, job) + } + return jobs, nil +} + // DeleteSendJob removes a row by id from the named queue. // If the queue is empty after deletion, the DB file is removed. func DeleteSendJob(storagePath, queue string, id int64) error {