ProcessSentMessages added for server acks
Some checks failed
continuous-integration/drone/push Build is failing
Some checks failed
continuous-integration/drone/push Build is failing
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user