From e0ad28516f8e37f669f77f2fc81cfa79fe48e93a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 14:52:04 -0400 Subject: [PATCH] Quote Signal replies from the v2 message on v2-primary A Signal quote-reply on a v2-primary daemon carried only the quoted message's v2 remote ID to the transport, which resolved the quote from the legacy store. That store lacks every send made through the v2 outbox, and holds phone-sent messages and some incoming ones under other IDs, so those replies failed with "signal reply target not found". The dispatcher now loads the quoted v2 message and fills bridge.MessageRef with its author, occurred time, body and first attachment, following the outbox from a pending request ID to the confirmed message. The Signal adapter passes the ref through, and signallive.QuoteArgs builds the quote from it, keeping the legacy lookup for undescribed refs and legacy IDs. Co-Authored-By: Claude Opus 5.5 --- cmd/r5_signal_quote_reply_test.go | 328 ++++++++++ docs/agent-runbook.md | 40 ++ internal/bridge/contracts.go | 18 +- .../signal/reply_target_test.go | 104 ++++ internal/bridgeadapters/signal/signal.go | 35 +- internal/bridgeadapters/signal/signal_test.go | 17 +- internal/messaging/dispatch.go | 49 +- internal/messaging/reply_ref.go | 154 +++++ internal/messaging/reply_ref_test.go | 583 ++++++++++++++++++ internal/messaging/types.go | 5 + internal/signallive/client.go | 43 +- internal/signallive/client_test.go | 16 +- internal/signallive/export_test.go | 35 ++ internal/signallive/reply_quote.go | 146 +++++ .../reply_quote_differential_external_test.go | 398 ++++++++++++ internal/signallive/reply_quote_test.go | 489 +++++++++++++++ internal/storage/sqlite/reply_target.go | 48 ++ internal/storage/sqlite/reply_target_test.go | 107 ++++ 18 files changed, 2534 insertions(+), 81 deletions(-) create mode 100644 cmd/r5_signal_quote_reply_test.go create mode 100644 internal/bridgeadapters/signal/reply_target_test.go create mode 100644 internal/messaging/reply_ref.go create mode 100644 internal/messaging/reply_ref_test.go create mode 100644 internal/signallive/reply_quote.go create mode 100644 internal/signallive/reply_quote_differential_external_test.go create mode 100644 internal/signallive/reply_quote_test.go create mode 100644 internal/storage/sqlite/reply_target.go create mode 100644 internal/storage/sqlite/reply_target_test.go diff --git a/cmd/r5_signal_quote_reply_test.go b/cmd/r5_signal_quote_reply_test.go new file mode 100644 index 00000000..22a5a572 --- /dev/null +++ b/cmd/r5_signal_quote_reply_test.go @@ -0,0 +1,328 @@ +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/rs/zerolog" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/bridgeadapters/scripted" + signaladapter "github.com/maxghenis/openmessage/internal/bridgeadapters/signal" + "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/ingest" + "github.com/maxghenis/openmessage/internal/messaging" + "github.com/maxghenis/openmessage/internal/signallive" + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" + "github.com/maxghenis/openmessage/internal/v2read" + "github.com/maxghenis/openmessage/internal/v2wire" + "github.com/maxghenis/openmessage/internal/web" +) + +const r5SignalAccountAddress = "+15550009999" + +// TestR5SignalQuoteRepliesResolveFromV2OnV2Primary runs Signal quote-replies +// end to end on a migrated v2-primary stack with a scripted Signal account: +// submission (HTTP API and v2wire), the durable dispatcher, the bridge +// request, the real Signal adapter conversion and signallive's quote builder. +// It covers the message the legacy store never holds on v2-primary (this +// account's own send through the v2 outbox), a live-ingested incoming message +// (SHA-1 remote ID), migrated messages both stores hold (quotes must equal +// what the legacy row gives), and a reply submitted while its quoted send was +// still pending. +func TestR5SignalQuoteRepliesResolveFromV2OnV2Primary(t *testing.T) { + fixture := buildR5LegacyFixture(t) + now := time.Date(2026, 7, 17, 18, 0, 0, 0, time.UTC) + runR5Migrate(t, fixture.DataDir, filepath.Join(fixture.DataDir, "v2"), false, now) + + t.Setenv("OPENMESSAGES_DATA_DIR", fixture.DataDir) + t.Setenv("OPENMESSAGES_DEMO", "0") + t.Setenv("OPENMESSAGES_APP_SANDBOX", "1") + t.Setenv("OPENMESSAGES_V2_PRIMARY", "1") + t.Setenv("OPENMESSAGES_V2_SEND", "") + t.Setenv("OPENMESSAGES_V2_INGEST", "") + + stack, err := newV2Stack(v2StackDeps{DataDir: fixture.DataDir, Logger: zerolog.Nop()}) + if err != nil { + t.Fatalf("open migrated v2 stack: %v", err) + } + t.Cleanup(func() { _ = stack.Store.Close() }) + signal := scripted.New(r5SignalAccountID, bridge.PlatformSignal) + if err := stack.RegisterAdapter(signal); err != nil { + t.Fatalf("register scripted Signal adapter: %v", err) + } + // The v2-primary daemon runs no legacy projector; this inert store is + // what it would write a confirmed send into. + inertLegacy, err := db.New(":memory:") + if err != nil { + t.Fatalf("create inert legacy store: %v", err) + } + t.Cleanup(func() { _ = inertLegacy.Close() }) + ctx, cancel := context.WithCancel(context.Background()) + stopStack := stack.Start(ctx, inertLegacy, nil, true) + t.Cleanup(func() { + stopStack() + cancel() + }) + + messages, err := sqlite.NewMessageRepository(stack.Store, time.Now) + if err != nil { + t.Fatalf("NewMessageRepository(): %v", err) + } + reads := v2read.New(stack.Store) + handler := web.APIHandlerWithOptions(inertLegacy, nil, zerolog.Nop(), nil, web.APIOptions{ + Reads: reads, + V2Primary: true, + V2: &web.V2Options{ + Service: stack.Service, Media: stack.Media, V2Store: stack.Store, + Blobs: stack.Blobs, Registry: stack.Registry, + }, + }) + deps := v2wire.NativeDeps{V2: stack.Store, Service: stack.Service, Registry: stack.Registry} + signalConversation := v2keys.DeriveID("conversation", r5SignalAccountID, r5SignalConversation) + + waitConfirmed := func(description, outboxID, remoteID string) { + t.Helper() + waitR5(t, description, func() bool { + delivery, getErr := stack.Service.Get(context.Background(), outboxID) + return getErr == nil && delivery.State == messaging.OutboxConfirmed && delivery.RemoteMessageID == remoteID + }) + } + lastTextReply := func() *bridge.MessageRef { + t.Helper() + requests := signal.TextRequests() + if len(requests) == 0 { + t.Fatal("no Signal text requests") + } + return requests[len(requests)-1].ReplyTo + } + quote := func(ref *bridge.MessageRef) []string { + t.Helper() + // An identity resolver is what resolveContactAddress does for an + // address with no known contact. + args, quoteErr := signallive.QuoteArgs(signaladapter.ReplyTarget(ref), r5SignalAccountAddress, func(value string) string { return value }) + if quoteErr != nil { + t.Fatalf("QuoteArgs(%+v): %v", ref, quoteErr) + } + return args + } + occurredAt := func(messageID string) time.Time { + t.Helper() + message, getErr := messages.GetMessage(context.Background(), messageID) + if getErr != nil { + t.Fatalf("GetMessage(%q): %v", messageID, getErr) + } + return time.UnixMilli(message.OccurredAtMS) + } + + t.Run("reply to this account's own v2 outbox send, which no legacy row holds", func(t *testing.T) { + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: "1752800000001"}) + own, err := v2wire.SubmitTextV2(context.Background(), deps, v2wire.TextInput{ + ConversationID: signalConversation, + Body: "r5 own outbox send", + IdempotencyKey: "r5-quote-own-send", + }) + if err != nil { + t.Fatalf("submit own send: %v", err) + } + waitConfirmed("own send", own.OutboxID, "1752800000001") + if rows, err := inertLegacy.GetMessagesByConversation(r5SignalConversation, 10); err != nil || len(rows) != 0 { + t.Fatalf("v2-primary wrote %d legacy rows (%v); the quoted send must exist only in v2", len(rows), err) + } + + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: "1752800000002"}) + body, _ := json.Marshal(map[string]string{ + "conversation_id": signalConversation, + "body": "r5 reply to own send", + "reply_to_id": own.LocalMessageID, + "idempotency_key": "r5-quote-own-reply", + }) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/api/v1/outbox/messages", bytes.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + handler.ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK { + t.Fatalf("POST reply = %d: %s", recorder.Code, recorder.Body.String()) + } + var submitted struct { + OutboxID string `json:"outbox_id"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &submitted); err != nil || submitted.OutboxID == "" { + t.Fatalf("decode reply submission %q: %v", recorder.Body.String(), err) + } + waitConfirmed("reply to own send", submitted.OutboxID, "1752800000002") + + ref := lastTextReply() + want := bridge.MessageRef{RemoteID: "1752800000001", SentAt: occurredAt(own.LocalMessageID), Text: "r5 own outbox send"} + if !r5MessageRefEqual(ref, want) { + t.Fatalf("reply ref = %+v, want %+v", ref, want) + } + // The quote carries Signal's timestamp for the send, not the submit + // time v2 stored as its occurred time. + if got, want := quote(ref), r5QuoteArgs("1752800000001", r5SignalAccountAddress, "r5 own outbox send"); !slices.Equal(got, want) { + t.Fatalf("quote = %q, want %q", got, want) + } + }) + + t.Run("media reply to a live-ingested incoming message", func(t *testing.T) { + const body = "r5 live incoming to quote" + timestamp := fixture.BaseMS + 40_000 + line := []byte(fmt.Sprintf( + `{"account":%q,"envelope":{"sourceServiceId":%q,"sourceName":"R5 Signal Quoted","timestamp":%d,"dataMessage":{"timestamp":%d,"message":%q}}}`, + r5SignalAccountAddress, r5SignalACI, timestamp, timestamp, body, + )) + record, ephemeral, err := ingest.BuildSignalIngress( + r5SignalAccountID, 1, r5SignalAccountAddress, line, "", "", time.UnixMilli(timestamp), + ) + if err != nil || record == nil || ephemeral != nil { + t.Fatalf("BuildSignalIngress = %v, %v, %v", record, ephemeral, err) + } + if err := stack.Sink.AppendIngress(context.Background(), *record); err != nil { + t.Fatalf("AppendIngress: %v", err) + } + var incoming *db.Message + waitR5(t, "incoming projection", func() bool { + found, searchErr := reads.SearchMessagesFiltered(body, db.SearchFilter{Limit: 5}) + if searchErr != nil || len(found) != 1 { + return false + } + incoming = found[0] + return true + }) + + signal.EnqueueMediaResult(bridge.SendResult{RemoteMessageID: "1752800000003"}) + submission, err := v2wire.SubmitMediaV2(context.Background(), deps, v2wire.MediaInput{ + ConversationID: signalConversation, + Content: strings.NewReader("r5 reply photo"), + Filename: "reply.png", + MIME: "image/png", + ReplyToID: incoming.MessageID, + IdempotencyKey: "r5-quote-incoming-media-reply", + }) + if err != nil { + t.Fatalf("submit media reply: %v", err) + } + waitConfirmed("media reply", submission.OutboxID, "1752800000003") + + calls := signal.MediaRequests() + ref := calls[len(calls)-1].Request.ReplyTo + want := bridge.MessageRef{ + RemoteID: v2keys.SignalIncomingSourceID(r5SignalConversation, r5SignalACI, timestamp), + AuthorID: r5SignalACI, + SentAt: time.UnixMilli(timestamp), + Text: body, + } + if !r5MessageRefEqual(ref, want) { + t.Fatalf("media reply ref = %+v, want %+v", ref, want) + } + if got, want := quote(ref), r5QuoteArgs(strconv.FormatInt(timestamp, 10), r5SignalACI, body); !slices.Equal(got, want) { + t.Fatalf("quote = %q, want %q", got, want) + } + }) + + t.Run("replies to migrated messages quote what their legacy rows give", func(t *testing.T) { + legacyFixture, err := db.New(fixture.StorePath) + if err != nil { + t.Fatalf("open legacy fixture store: %v", err) + } + defer legacyFixture.Close() + for index, migrated := range []struct{ legacyID, remoteID string }{ + {legacyID: "signal:1700000001000", remoteID: "1700000001000"}, + {legacyID: "signal:local:abc123r5", remoteID: "local:abc123r5"}, + } { + row, err := legacyFixture.GetMessageByID(migrated.legacyID) + if err != nil || row == nil { + t.Fatalf("legacy fixture row %q = %v, %v", migrated.legacyID, row, err) + } + v2ID := v2keys.DeriveID("message", r5SignalAccountID, r5SignalConversation+"\x1f"+migrated.remoteID) + result := strconv.Itoa(1752800000010 + index) + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: result}) + submission, err := v2wire.SubmitTextV2(context.Background(), deps, v2wire.TextInput{ + ConversationID: signalConversation, + Body: "r5 reply to migrated " + migrated.remoteID, + ReplyToID: v2ID, + IdempotencyKey: "r5-quote-migrated-" + migrated.remoteID, + }) + if err != nil { + t.Fatalf("submit reply to %q: %v", v2ID, err) + } + waitConfirmed("reply to "+migrated.legacyID, submission.OutboxID, result) + + // The legacy lookup's rules: the row's own timestamp, this account + // for a row it sent, otherwise the stored sender. + author := row.SenderNumber + if row.IsFromMe { + author = r5SignalAccountAddress + } + want := r5QuoteArgs(strconv.FormatInt(row.TimestampMS, 10), author, row.Body) + if got := quote(lastTextReply()); !slices.Equal(got, want) { + t.Fatalf("quote for migrated %q = %q, legacy row gives %q", migrated.legacyID, got, want) + } + } + }) + + t.Run("reply submitted while its quoted send was still pending", func(t *testing.T) { + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: "1752800000020"}) + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: "1752800000021"}) + start := time.Now() + quoted, err := v2wire.SubmitTextV2(context.Background(), deps, v2wire.TextInput{ + ConversationID: signalConversation, + Body: "r5 scheduled quoted send", + IdempotencyKey: "r5-quote-pending-quoted", + NotBefore: start.Add(1500 * time.Millisecond), + }) + if err != nil { + t.Fatalf("submit scheduled quoted send: %v", err) + } + reply, err := v2wire.SubmitTextV2(context.Background(), deps, v2wire.TextInput{ + ConversationID: signalConversation, + Body: "r5 reply queued behind it", + ReplyToID: quoted.LocalMessageID, + IdempotencyKey: "r5-quote-pending-reply", + NotBefore: start.Add(2500 * time.Millisecond), + }) + if err != nil { + t.Fatalf("submit reply: %v", err) + } + stored, err := messages.GetMessage(context.Background(), reply.LocalMessageID) + if err != nil || stored.ReplyToRemoteID == nil || *stored.ReplyToRemoteID == "1752800000020" { + t.Fatalf("reply stored target %v (%v), want the quoted send's pending request ID", stored.ReplyToRemoteID, err) + } + waitConfirmed("quoted send", quoted.OutboxID, "1752800000020") + waitConfirmed("reply", reply.OutboxID, "1752800000021") + + ref := lastTextReply() + want := bridge.MessageRef{RemoteID: "1752800000020", SentAt: occurredAt(quoted.LocalMessageID), Text: "r5 scheduled quoted send"} + if !r5MessageRefEqual(ref, want) { + t.Fatalf("reply ref = %+v, want %+v", ref, want) + } + if got, want := quote(ref), r5QuoteArgs("1752800000020", r5SignalAccountAddress, "r5 scheduled quoted send"); !slices.Equal(got, want) { + t.Fatalf("quote = %q, want %q", got, want) + } + }) +} + +func r5QuoteArgs(timestamp, author, message string) []string { + return []string{"--quote-timestamp", timestamp, "--quote-author", author, "--quote-message", message} +} + +func r5MessageRefEqual(got *bridge.MessageRef, want bridge.MessageRef) bool { + return got != nil && + got.RemoteID == want.RemoteID && + got.AuthorID == want.AuthorID && + got.SentAt.Equal(want.SentAt) && + got.Text == want.Text && + got.HasAttachment == want.HasAttachment && + got.AttachmentMIME == want.AttachmentMIME +} diff --git a/docs/agent-runbook.md b/docs/agent-runbook.md index 127beb70..fcd3c4d0 100644 --- a/docs/agent-runbook.md +++ b/docs/agent-runbook.md @@ -1464,6 +1464,46 @@ up the whole `signal-cli/` dir first and restore the media subdirs after the new link (recipe proven 2026-07-20; backups live under `/app-backups/`). +### Signal quote-replies on v2-primary + +Signal quotes a message by its author and sent timestamp, passed to signal-cli +as `--quote-timestamp/--quote-author/--quote-message`. The outbox stores only +the quoted message's v2 remote ID, which for an incoming message is a SHA-1 +(`v2keys.SignalIncomingSourceID`), not a timestamp. So the dispatcher +(`messaging.replyRefForLease`) loads the quoted v2 message and puts its author, +`occurred_at_ms`, body and first attachment on the request's +`bridge.MessageRef`, and `signallive.QuoteArgs` builds the quote from those. +No legacy row is needed. Before this, the Signal transport looked the quote up +in the legacy `messages.db` only (the retained receive handler still writes +it on v2-primary), and on v2-primary that lookup cannot find: + +- any send this account made through the v2 outbox. The legacy projector runs + only on legacy-primary, and the adapter expects no echo of its own sends; +- messages this account sent from the phone after cutover. Legacy keeps them + as `signal:local:`, v2 as the bare timestamp; +- incoming messages whose sender the legacy handler resolved differently from + the capture-time contact cache (it refreshes contacts on a miss, capture + does not), which puts them under different IDs. + +Rules worth knowing when a quote looks wrong: + +- The timestamp is `occurred_at_ms` (the decoder and the migration both set it + to the Signal sent timestamp), except for a message this account wrote whose + remote ID is a decimal timestamp. A send through the outbox keeps its + *submit* time as `occurred_at_ms`, and confirming it sets the remote ID to the + timestamp signal-cli reported, so that ID wins. +- A reply submitted while its quoted send was still pending stored the quoted + message's transport request ID. The dispatcher follows the outbox row from + that ID to the message's current remote ID. +- A quoted message v2 does not hold, or an outgoing one still awaiting its + transport ID, gets a bare ref, and the transport falls back to the legacy + lookup. A delivery that fails with `signal reply target not found` is that + case. Check the quoted message's own outbox row + (`GET /api/v1/outbox/`). +- On legacy-primary with `OPENMESSAGES_V2_SEND=1`, the mirror keys a reply + target by its legacy ID (`signal:`) and records no sender or attachment. + Such an ID still quotes from its legacy row when that row exists. + ## Deploying a new build to a live install **`RELEASE=1` is required.** Without it `build.sh` stamps the dev bundle id diff --git a/internal/bridge/contracts.go b/internal/bridge/contracts.go index f542acf4..6a85eb05 100644 --- a/internal/bridge/contracts.go +++ b/internal/bridge/contracts.go @@ -60,11 +60,27 @@ type ConversationRef struct { RemoteID string } +// MessageRef names a stored message an operation refers to: a reaction or +// read-receipt target, or the message a reply quotes. RemoteID is the +// transport's identity for it. The dispatcher fills the other fields from the +// stored message. AuthorID is the author's canonical identity. It is empty +// when the stored message names no sender, which is always the case for a +// message this account sent, and adapters read empty as this account. SentAt +// is the message's occurred time. A zero +// SentAt means the dispatcher had no stored message to describe (it does not +// hold one under RemoteID, or the message is an outgoing one still waiting +// for its transport ID), so only RemoteID is meaningful. type MessageRef struct { RemoteID string AuthorID string SentAt time.Time - Text string + // Text, HasAttachment and AttachmentMIME are filled for reply targets + // only. Text is the stored body. HasAttachment reports whether the message + // carries an attachment, and AttachmentMIME is the first attachment's MIME + // type, so an adapter can describe a body-less message in its quote. + Text string + HasAttachment bool + AttachmentMIME string } type SendResult struct { diff --git a/internal/bridgeadapters/signal/reply_target_test.go b/internal/bridgeadapters/signal/reply_target_test.go new file mode 100644 index 00000000..925757d1 --- /dev/null +++ b/internal/bridgeadapters/signal/reply_target_test.go @@ -0,0 +1,104 @@ +package signal + +import ( + "bytes" + "context" + "reflect" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/signallive" +) + +func TestReplyTargetCarriesEveryMessageRefField(t *testing.T) { + if got := ReplyTarget(nil); got != (signallive.ReplyTarget{}) { + t.Fatalf("ReplyTarget(nil) = %+v, want no reply", got) + } + + ref := bridge.MessageRef{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_123), + Text: "quoted", + HasAttachment: true, + AttachmentMIME: "image/jpeg", + } + got := reflect.ValueOf(ReplyTarget(&ref)) + source := reflect.ValueOf(ref) + // Every MessageRef field must reach signallive under the same name, so a + // field added to the contract cannot be dropped here silently. + for index := 0; index < source.NumField(); index++ { + name := source.Type().Field(index).Name + field := got.FieldByName(name) + if !field.IsValid() { + t.Fatalf("signallive.ReplyTarget has no %s field", name) + } + if !reflect.DeepEqual(field.Interface(), source.Field(index).Interface()) { + t.Fatalf("ReplyTarget().%s = %v, want %v", name, field.Interface(), source.Field(index).Interface()) + } + if source.Field(index).IsZero() { + t.Fatalf("fixture leaves MessageRef.%s zero; set it so the copy is checked", name) + } + } + if got.NumField() != source.NumField() { + t.Fatalf("ReplyTarget has %d fields, MessageRef %d", got.NumField(), source.NumField()) + } +} + +func TestSendTextAndMediaHandTheDescribedReplyToThePoller(t *testing.T) { + poller := newFakePoller() + poller.status = signallive.StatusSnapshot{Connected: true, Paired: true, Account: "+15551230000"} + poller.textTimestamp = 1700000000124 + poller.mediaTimestamp = 1700000000125 + adapter := &Adapter{accountID: "signal-primary", poller: poller} + reply := &bridge.MessageRef{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(1_700_000_000_000), + Text: "sent through the v2 outbox", + } + want := signallive.ReplyTarget{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(1_700_000_000_000), + Text: "sent through the v2 outbox", + } + + if _, err := adapter.SendText(context.Background(), bridge.TextRequest{ + AccountID: "signal-primary", + Conversation: bridge.ConversationRef{RemoteID: "signal:+15551234567"}, + Body: "replying", + ReplyTo: reply, + }); err != nil { + t.Fatalf("SendText(): %v", err) + } + if got := poller.lastTextRequest().reply; got != want { + t.Fatalf("text reply handed to signallive = %+v, want %+v", got, want) + } + + content := []byte("png") + if _, err := adapter.SendMedia(context.Background(), bridge.MediaRequest{ + AccountID: "signal-primary", + Conversation: bridge.ConversationRef{RemoteID: "signal:+15551234567"}, + Reader: bytes.NewReader(content), + Size: int64(len(content)), + Filename: "reply.png", + MIME: "image/png", + ReplyTo: reply, + }); err != nil { + t.Fatalf("SendMedia(): %v", err) + } + if got := poller.lastMediaRequest().reply; got != want { + t.Fatalf("media reply handed to signallive = %+v, want %+v", got, want) + } + + if _, err := adapter.SendText(context.Background(), bridge.TextRequest{ + AccountID: "signal-primary", + Conversation: bridge.ConversationRef{RemoteID: "signal:+15551234567"}, + Body: "no reply", + }); err != nil { + t.Fatalf("SendText(no reply): %v", err) + } + if got := poller.lastTextRequest().reply; got != (signallive.ReplyTarget{}) { + t.Fatalf("text without a reply handed %+v, want none", got) + } +} diff --git a/internal/bridgeadapters/signal/signal.go b/internal/bridgeadapters/signal/signal.go index 98d4f892..d5ce987c 100644 --- a/internal/bridgeadapters/signal/signal.go +++ b/internal/bridgeadapters/signal/signal.go @@ -25,9 +25,9 @@ import ( type poller interface { ObserveIngress(func(account string, line []byte, resolvedSource string, resolvedDestination string)) func() StartPoller(context.Context) (signallive.PollerRun, error) - SendTextRequest(string, string, string) (int64, error) + SendTextRequest(string, string, signallive.ReplyTarget) (int64, error) SendReactionRequest(string, string, string, string, string) error - SendMediaRequest(string, io.Reader, int64, string, string, string, string) (int64, error) + SendMediaRequest(string, io.Reader, int64, string, string, string, signallive.ReplyTarget) (int64, error) DownloadMediaRef(string, string, string, string, string, bool) ([]byte, string, error) Status() signallive.StatusSnapshot ApplyPollerFailure(signallive.PollerExit) @@ -126,14 +126,10 @@ func (a *Adapter) SendText( return bridge.SendResult{}, signalNotConnectedError("send_text") } - replyToID := "" - if req.ReplyTo != nil { - replyToID = req.ReplyTo.RemoteID - } timestampMS, err := a.poller.SendTextRequest( req.Conversation.RemoteID, req.Body, - replyToID, + ReplyTarget(req.ReplyTo), ) if err != nil { a.ReportError(err) @@ -155,6 +151,25 @@ func (a *Adapter) SendText( }, nil } +// ReplyTarget carries a durable reply reference to signallive unchanged. The +// dispatcher's description of the quoted message (author, occurred time, +// body, first attachment) lets signallive quote it without a legacy row; a +// reference without one (zero SentAt) is still resolved from the legacy store +// by RemoteID. A nil reference is no reply. +func ReplyTarget(ref *bridge.MessageRef) signallive.ReplyTarget { + if ref == nil { + return signallive.ReplyTarget{} + } + return signallive.ReplyTarget{ + RemoteID: ref.RemoteID, + AuthorID: ref.AuthorID, + SentAt: ref.SentAt, + Text: ref.Text, + HasAttachment: ref.HasAttachment, + AttachmentMIME: ref.AttachmentMIME, + } +} + // SendReaction adapts a durable reaction request to Signal's store-free // signal-cli path. Reactions intentionally confirm without a remote result ID. func (a *Adapter) SendReaction( @@ -241,10 +256,6 @@ func (a *Adapter) SendMedia( return bridge.SendResult{}, signalNotConnectedError("send_media") } - replyToID := "" - if req.ReplyTo != nil { - replyToID = req.ReplyTo.RemoteID - } timestampMS, err := a.poller.SendMediaRequest( req.Conversation.RemoteID, req.Reader, @@ -252,7 +263,7 @@ func (a *Adapter) SendMedia( req.Filename, req.MIME, req.Caption, - replyToID, + ReplyTarget(req.ReplyTo), ) if err != nil { a.ReportError(err) diff --git a/internal/bridgeadapters/signal/signal_test.go b/internal/bridgeadapters/signal/signal_test.go index 1f665c7f..a1784de6 100644 --- a/internal/bridgeadapters/signal/signal_test.go +++ b/internal/bridgeadapters/signal/signal_test.go @@ -166,7 +166,7 @@ func TestSendTextMapsRetainedTimestampAndRequest(t *testing.T) { request := poller.lastTextRequest() if request.conversationID != "signal:+15551234567" || request.body != "hello from durable Signal" || - request.replyToID != "signal:1700000000000" { + request.reply != (signallive.ReplyTarget{RemoteID: "signal:1700000000000"}) { t.Fatalf("retained SendTextRequest = %+v, want mapped TextRequest", request) } } @@ -679,7 +679,7 @@ func TestSendMediaMapsRetainedTimestampAndRequest(t *testing.T) { request.filename != "photo.png" || request.mime != "image/png" || request.caption != "signal photo" || - request.replyToID != "signal:1700000000000" || + request.reply != (signallive.ReplyTarget{RemoteID: "signal:1700000000000"}) || !bytes.Equal(request.content, content) { t.Fatalf("retained SendMediaRequest = %+v, want mapped MediaRequest", request) } @@ -1525,7 +1525,7 @@ type fakePoller struct { type fakeTextRequest struct { conversationID string body string - replyToID string + reply signallive.ReplyTarget } type fakeReactionRequest struct { @@ -1543,7 +1543,7 @@ type fakeMediaRequest struct { filename string mime string caption string - replyToID string + reply signallive.ReplyTarget } type fakeDownloadRequest struct { @@ -1653,12 +1653,12 @@ func (p *fakePoller) Status() signallive.StatusSnapshot { return p.status } -func (p *fakePoller) SendTextRequest(conversationID, body, replyToID string) (int64, error) { +func (p *fakePoller) SendTextRequest(conversationID, body string, reply signallive.ReplyTarget) (int64, error) { p.mu.Lock() p.textCalls = append(p.textCalls, fakeTextRequest{ conversationID: conversationID, body: body, - replyToID: replyToID, + reply: reply, }) timestamp, err := p.textTimestamp, p.textErr p.mu.Unlock() @@ -1715,7 +1715,8 @@ func (p *fakePoller) SendMediaRequest( conversationID string, content io.Reader, size int64, - filename, mime, caption, replyToID string, + filename, mime, caption string, + reply signallive.ReplyTarget, ) (int64, error) { data, readErr := io.ReadAll(content) p.mu.Lock() @@ -1726,7 +1727,7 @@ func (p *fakePoller) SendMediaRequest( filename: filename, mime: mime, caption: caption, - replyToID: replyToID, + reply: reply, }) timestamp, sendErr := p.mediaTimestamp, p.mediaErr p.mu.Unlock() diff --git a/internal/messaging/dispatch.go b/internal/messaging/dispatch.go index afa99791..2adca45c 100644 --- a/internal/messaging/dispatch.go +++ b/internal/messaging/dispatch.go @@ -185,6 +185,17 @@ func (s *MessageService) dispatchTextLease(ctx context.Context, outboxLease sqli s.clock.Now().Add(s.retryDelay), ) } + replyTo, code, err := s.replyRefForLease(ctx, item, message) + if err != nil { + return s.markPreCallFailure( + ctx, + item, + string(bridge.FailureTransient), + code, + err, + s.clock.Now().Add(s.retryDelay), + ) + } if err := s.outbox.MarkTransportCalled(ctx, sqlite.Attempt{ OutboxID: item.OutboxID, @@ -208,9 +219,7 @@ func (s *MessageService) dispatchTextLease(ctx context.Context, outboxLease sqli Body: message.Body, RequestID: item.TransportRequestID, } - if message.ReplyToRemoteID != nil { - request.ReplyTo = &bridge.MessageRef{RemoteID: *message.ReplyToRemoteID} - } + request.ReplyTo = replyTo result, sendErr := dispatchLease.Text.SendText(ctx, request) mutationCtx, cancel := s.storageMutationContext(ctx) defer cancel() @@ -323,6 +332,17 @@ func (s *MessageService) dispatchMediaLease(ctx context.Context, outboxLease sql s.clock.Now().Add(s.retryDelay), ) } + replyTo, code, err := s.replyRefForLease(ctx, item, message) + if err != nil { + return s.markPreCallFailure( + ctx, + item, + string(bridge.FailureTransient), + code, + err, + s.clock.Now().Add(s.retryDelay), + ) + } attachment, err := s.outbox.GetOutboxAttachment(ctx, item.OutboxID) if err != nil { @@ -383,9 +403,7 @@ func (s *MessageService) dispatchMediaLease(ctx context.Context, outboxLease sql Caption: message.Body, RequestID: item.TransportRequestID, } - if message.ReplyToRemoteID != nil { - request.ReplyTo = &bridge.MessageRef{RemoteID: *message.ReplyToRemoteID} - } + request.ReplyTo = replyTo result, sendErr := dispatchLease.Media.SendMedia(ctx, request) mutationCtx, cancel := s.storageMutationContext(ctx) defer cancel() @@ -717,23 +735,10 @@ func (s *MessageService) targetRefForLease( ) } - authorID := "" - if target.SenderIdentityID != nil { - identity, err := s.store.GetIdentity(*target.SenderIdentityID) - if err != nil { - return bridge.MessageRef{}, "load_target_author", err - } - if identity.AccountID != item.AccountID { - return bridge.MessageRef{}, "load_target_author", fmt.Errorf( - "target author %q belongs to account %q", - identity.IdentityID, - identity.AccountID, - ) - } - authorID = identity.CanonicalValue + authorID, err := s.messageAuthorID(item, target) + if err != nil { + return bridge.MessageRef{}, "load_target_author", err } - // A nil sender marks an outgoing message authored by self, so an empty - // transport-neutral AuthorID preserves that distinction for adapter shims. // Until echo reconciliation, an outgoing target's RemoteID remains its // stable transport request ID. return bridge.MessageRef{ diff --git a/internal/messaging/reply_ref.go b/internal/messaging/reply_ref.go new file mode 100644 index 00000000..c8dd3dd8 --- /dev/null +++ b/internal/messaging/reply_ref.go @@ -0,0 +1,154 @@ +package messaging + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +// replyRefForLease describes the message a text or media send replies to. +// +// The outbox message stores only the quoted message's remote ID, as it stood +// when the reply was submitted. A transport that quotes by author and sent +// time rather than by remote ID (Signal) needs the stored message itself, so +// the ref carries its author, occurred time, body and first attachment. +// +// The quoted message is the one the conversation holds under that remote ID. +// When the reply was submitted while the quoted message was an outgoing one +// still waiting for its transport ID, the stored ID is that message's +// transport request ID. Confirming the send has since moved the message to +// the transport's ID, so the outbox row carrying the request ID leads back to +// it, and the ref names its current remote ID. +// +// A quoted message the store does not hold, or an outgoing one the transport +// has not accepted yet, yields a bare ref (RemoteID only), which is what every +// reply carried before. Store errors fail the attempt before the transport is +// called, like the other pre-call loads. +func (s *MessageService) replyRefForLease( + ctx context.Context, + item sqlite.OutboxItem, + message sqlite.Message, +) (*bridge.MessageRef, string, error) { + if message.ReplyToRemoteID == nil { + return nil, "", nil + } + remoteID := *message.ReplyToRemoteID + target, found, err := s.replyTarget(ctx, item, remoteID) + if err != nil { + return nil, "load_reply_target", err + } + if !found { + return &bridge.MessageRef{RemoteID: remoteID}, "", nil + } + awaiting, err := s.awaitingTransportID(ctx, item, target) + if err != nil { + return nil, "load_reply_target", err + } + if awaiting { + return &bridge.MessageRef{RemoteID: target.RemoteMessageID}, "", nil + } + authorID, err := s.messageAuthorID(item, target) + if err != nil { + return nil, "load_reply_target_author", err + } + mime, hasAttachment, err := s.messages.FirstAttachmentMIME(ctx, target.MessageID) + if err != nil { + return nil, "load_reply_target_attachment", err + } + return &bridge.MessageRef{ + RemoteID: target.RemoteMessageID, + AuthorID: authorID, + SentAt: time.UnixMilli(target.OccurredAtMS), + Text: target.Body, + HasAttachment: hasAttachment, + AttachmentMIME: mime, + }, "", nil +} + +// replyTarget finds the message a reply's stored remote ID names in the +// outbox item's conversation: the message held under that remote ID, else the +// local message of the outbox row whose transport request ID it is. +func (s *MessageService) replyTarget( + ctx context.Context, + item sqlite.OutboxItem, + remoteID string, +) (sqlite.Message, bool, error) { + target, err := s.messages.GetMessageByRemote(ctx, item.AccountID, item.ConversationID, remoteID) + if err == nil { + return target, true, nil + } + if !errors.Is(err, sqlite.ErrNotFound) { + return sqlite.Message{}, false, err + } + + row, err := s.outbox.FindByTransportRequestID(ctx, item.AccountID, remoteID) + if errors.Is(err, sqlite.ErrNotFound) { + return sqlite.Message{}, false, nil + } + if err != nil { + return sqlite.Message{}, false, err + } + if row.LocalMessageID == nil || row.ConversationID != item.ConversationID { + return sqlite.Message{}, false, nil + } + target, err = s.messages.GetMessage(ctx, *row.LocalMessageID) + if errors.Is(err, sqlite.ErrNotFound) { + return sqlite.Message{}, false, nil + } + if err != nil { + return sqlite.Message{}, false, err + } + if target.AccountID != item.AccountID || target.ConversationID != item.ConversationID { + return sqlite.Message{}, false, nil + } + return target, true, nil +} + +// awaitingTransportID reports whether target is an outgoing message whose +// remote ID is still the transport request ID of its own outbox row, so the +// transport has not given it an ID (Signal: a sent timestamp) yet. +func (s *MessageService) awaitingTransportID( + ctx context.Context, + item sqlite.OutboxItem, + target sqlite.Message, +) (bool, error) { + if target.Direction != sqlite.MessageDirectionOutgoing { + return false, nil + } + row, err := s.outbox.FindByTransportRequestID(ctx, item.AccountID, target.RemoteMessageID) + if errors.Is(err, sqlite.ErrNotFound) { + return false, nil + } + if err != nil { + return false, err + } + return row.LocalMessageID != nil && *row.LocalMessageID == target.MessageID, nil +} + +// messageAuthorID returns the canonical identity of a stored message's +// author. A nil sender marks an outgoing message authored by self, so an empty +// transport-neutral AuthorID preserves that distinction for adapter shims. +func (s *MessageService) messageAuthorID( + item sqlite.OutboxItem, + target sqlite.Message, +) (string, error) { + if target.SenderIdentityID == nil { + return "", nil + } + identity, err := s.store.GetIdentity(*target.SenderIdentityID) + if err != nil { + return "", err + } + if identity.AccountID != item.AccountID { + return "", fmt.Errorf( + "target author %q belongs to account %q", + identity.IdentityID, + identity.AccountID, + ) + } + return identity.CanonicalValue, nil +} diff --git a/internal/messaging/reply_ref_test.go b/internal/messaging/reply_ref_test.go new file mode 100644 index 00000000..fbc899b7 --- /dev/null +++ b/internal/messaging/reply_ref_test.go @@ -0,0 +1,583 @@ +package messaging + +import ( + "bytes" + "context" + "errors" + "fmt" + "math/rand" + "strings" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +func TestReplyToIncomingMessageCarriesItsStoredDescription(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + identityID := "identity-quoted-author" + seedDispatchIdentity(t, store, identityID, "+15551234567", clock.Now()) + target := projectReplyTarget(t, store, clock, sqlite.Message{ + MessageID: "message-quoted-incoming", + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: "4f2b7c9d0e1a3b5c7d9e1f2a4b6c8d0e2f4a6b8c", + SenderIdentityID: &identityID, + Direction: sqlite.MessageDirectionIncoming, + Body: " look at this ", + State: sqlite.MessageStateActive, + OccurredAtMS: 1_700_000_000_123, + }, "video/mp4", "image/jpeg") + + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "reply-text"}}}} + media := &scriptedMediaSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "reply-media"}}}} + registry := newScriptedRegistry("reply-incoming", text) + registry.setMediaSender(media) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-incoming-text"), + Body: "replying in text", + ReplyToMessageID: target.MessageID, + }) + mustSendMedia(t, service, SendMediaCommand{ + CommonCommand: testCommonCommand("reply-incoming-media"), + Content: bytes.NewReader([]byte("reply photo")), + Filename: "reply.png", + MIME: "image/png", + ReplyToMessageID: target.MessageID, + }) + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 2 { + t.Fatalf("DispatchDue() = %d, %v; want 2, nil", processed, err) + } + + want := bridge.MessageRef{ + RemoteID: target.RemoteMessageID, + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_123), + Text: " look at this ", + HasAttachment: true, + AttachmentMIME: "video/mp4", + } + textRequests := text.snapshotRequests() + if len(textRequests) != 1 || !messageRefEqual(textRequests[0].ReplyTo, &want) { + t.Fatalf("text reply ref = %+v, want %+v", textRequests, want) + } + mediaRequests := media.snapshotRequests() + if len(mediaRequests) != 1 || !messageRefEqual(mediaRequests[0].ReplyTo, &want) { + t.Fatalf("media reply ref = %+v, want %+v", mediaRequests, want) + } +} + +func TestReplyToOwnConfirmedSendCarriesTransportIDAndSelfAuthor(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{ + {result: bridge.SendResult{RemoteMessageID: "1700000000555"}}, + {result: bridge.SendResult{RemoteMessageID: "1700000000556"}}, + }} + registry := newScriptedRegistry("reply-own", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + quoted := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-own-quoted"), + Body: "sent through the outbox", + }) + submittedAt := clock.Now() + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(quoted) = %d, %v; want 1, nil", processed, err) + } + clock.Advance(time.Minute) + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-own-reply"), + Body: "replying to myself", + ReplyToMessageID: quoted.LocalMessageID, + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reply) = %d, %v; want 1, nil", processed, err) + } + + requests := text.snapshotRequests() + want := bridge.MessageRef{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(submittedAt.UnixMilli()), + Text: "sent through the outbox", + } + if len(requests) != 2 || !messageRefEqual(requests[1].ReplyTo, &want) { + t.Fatalf("own reply ref = %+v, want %+v", requests, want) + } +} + +func TestReplySubmittedWhileQuotedSendWasPendingFollowsItsConfirmation(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{ + {result: bridge.SendResult{RemoteMessageID: "1700000000777"}}, + {result: bridge.SendResult{RemoteMessageID: "1700000000778"}}, + }} + registry := newScriptedRegistry("reply-pending", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + quoted := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-pending-quoted"), + Body: "still sending", + }) + quotedRow := mustOutboxItem(t, service, quoted.OutboxID) + replyCommand := testCommonCommand("reply-pending-reply") + replyCommand.NotBefore = clock.Now().Add(time.Second) + reply := mustSendText(t, service, SendTextCommand{ + CommonCommand: replyCommand, + Body: "quick reply", + ReplyToMessageID: quoted.LocalMessageID, + }) + stored, err := service.messages.GetMessage(context.Background(), reply.LocalMessageID) + if err != nil { + t.Fatalf("GetMessage(reply): %v", err) + } + if stored.ReplyToRemoteID == nil || *stored.ReplyToRemoteID != quotedRow.TransportRequestID { + t.Fatalf("stored reply target = %v, want the pending request ID %q", stored.ReplyToRemoteID, quotedRow.TransportRequestID) + } + + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(quoted) = %d, %v; want 1, nil", processed, err) + } + clock.Advance(time.Second) + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reply) = %d, %v; want 1, nil", processed, err) + } + + requests := text.snapshotRequests() + want := bridge.MessageRef{ + RemoteID: "1700000000777", + SentAt: time.UnixMilli(messagingTestTime.UnixMilli()), + Text: "still sending", + } + if len(requests) != 2 || !messageRefEqual(requests[1].ReplyTo, &want) { + t.Fatalf("reply ref after the quoted send confirmed = %+v, want %+v", requests, want) + } +} + +func TestReplyToUnsentOwnMessageCarriesOnlyItsRequestID(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "reply-after-cancel"}}}} + registry := newScriptedRegistry("reply-unsent", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + quoted := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-unsent-quoted"), + Body: "never sent", + }) + quotedRow := mustOutboxItem(t, service, quoted.OutboxID) + if _, err := service.Cancel(context.Background(), quoted.OutboxID); err != nil { + t.Fatalf("Cancel(quoted): %v", err) + } + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-unsent-reply"), + Body: "reply to a canceled send", + ReplyToMessageID: quoted.LocalMessageID, + }) + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue() = %d, %v; want 1, nil", processed, err) + } + + requests := text.snapshotRequests() + want := bridge.MessageRef{RemoteID: quotedRow.TransportRequestID} + if len(requests) != 1 || !messageRefEqual(requests[0].ReplyTo, &want) { + t.Fatalf("reply ref to an unsent message = %+v, want bare %+v", requests, want) + } +} + +func TestReplyToRemoteIDTheStoreLacksStaysBare(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "reply-unknown"}}}} + registry := newScriptedRegistry("reply-unknown", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + unknown := "remote-id-the-store-does-not-hold" + if _, _, err := service.outbox.EnqueueOutgoingMessage(context.Background(), sqlite.NewOutboxItem{ + OutboxID: "outbox-unknown-reply", + AccountID: "account-1", + ConversationID: "conversation-1", + Kind: sqlite.OutboxKindText, + IdempotencyKey: "reply-unknown", + PayloadHash: "hash-unknown-reply", + Operation: textOperation, + LocalMessageID: "message-unknown-reply", + TransportRequestID: "request-unknown-reply", + ScheduledFor: clock.Now(), + }, sqlite.Message{ + MessageID: "message-unknown-reply", + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: "request-unknown-reply", + Direction: sqlite.MessageDirectionOutgoing, + Body: "reply to a message only the transport knows", + ReplyToRemoteID: &unknown, + State: sqlite.MessageStateActive, + OccurredAtMS: clock.Now().UnixMilli(), + }); err != nil { + t.Fatalf("EnqueueOutgoingMessage(): %v", err) + } + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue() = %d, %v; want 1, nil", processed, err) + } + + requests := text.snapshotRequests() + want := bridge.MessageRef{RemoteID: unknown} + if len(requests) != 1 || !messageRefEqual(requests[0].ReplyTo, &want) { + t.Fatalf("reply ref to an unheld remote ID = %+v, want bare %+v", requests, want) + } +} + +func TestReplyToOutboxMediaWithoutCaptionCarriesItsAttachment(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "reply-to-photo"}}}} + media := &scriptedMediaSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "1700000000999"}}}} + registry := newScriptedRegistry("reply-outbox-media", text) + registry.setMediaSender(media) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + photo := mustSendMedia(t, service, SendMediaCommand{ + CommonCommand: testCommonCommand("reply-outbox-media-photo"), + Content: bytes.NewReader([]byte("png bytes")), + Filename: "photo.png", + MIME: "image/png", + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(photo) = %d, %v; want 1, nil", processed, err) + } + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-outbox-media-reply"), + Body: "nice photo", + ReplyToMessageID: photo.LocalMessageID, + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reply) = %d, %v; want 1, nil", processed, err) + } + + requests := text.snapshotRequests() + want := bridge.MessageRef{ + RemoteID: "1700000000999", + SentAt: time.UnixMilli(messagingTestTime.UnixMilli()), + HasAttachment: true, + AttachmentMIME: "image/png", + } + if len(requests) != 1 || !messageRefEqual(requests[0].ReplyTo, &want) { + t.Fatalf("reply ref to an outbox photo = %+v, want %+v", requests, want) + } +} + +func TestReplyTargetLoadFailureFailsBeforeTransport(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + target := projectReplyTarget(t, store, clock, sqlite.Message{ + MessageID: "message-load-failure", + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: "remote-load-failure", + Direction: sqlite.MessageDirectionIncoming, + Body: "quoted", + State: sqlite.MessageStateActive, + OccurredAtMS: clock.Now().Add(-time.Minute).UnixMilli(), + }) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "never"}}}} + registry := newScriptedRegistry("reply-load-failure", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + service.messages = failingAttachmentRepository{messageRepository: service.messages} + + submission := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-load-failure"), + Body: "reply", + ReplyToMessageID: target.MessageID, + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue() = %d, %v; want 1, nil", processed, err) + } + row := mustOutboxItem(t, service, submission.OutboxID) + if row.State != sqlite.OutboxNotDispatched || row.ErrorCode == nil || + *row.ErrorCode != "load_reply_target_attachment" || row.TransportCalledAtMS != nil { + t.Fatalf("outbox row after reply-target load failure = %+v, want not_dispatched before the transport call", row) + } + if got := text.requestCount(); got != 0 { + t.Fatalf("transport calls = %d, want 0", got) + } +} + +// TestReplyRefInvariants checks, over seeded random conversations, every +// reply the dispatcher maps: +// +// - The ref names the quoted message's current remote ID, or the stored +// remote ID when the store has no such message. +// - SentAt is zero exactly when the dispatcher has no description: the +// message is not held, or it is an outgoing one the transport has not +// accepted. +// - A described ref equals the stored message: AuthorID is its sender's +// canonical value ("" for a nil sender), SentAt its occurred time, Text +// its body, and the attachment fields its first attachment. +// - The mapping is deterministic: a retried attempt carries the same ref. +func TestReplyRefInvariants(t *testing.T) { + for seed := int64(1); seed <= 40; seed++ { + seed := seed + t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { + checkReplyRefInvariants(t, seed) + }) + } +} + +type replyRefCase struct { + name string + quotedID string // local message ID the reply quotes + wantRemote string + described bool + author string + occurredMS int64 + body string + mime string + hasMedia bool +} + +func checkReplyRefInvariants(t *testing.T, seed int64) { + random := rand.New(rand.NewSource(seed)) + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{} + media := &scriptedMediaSender{} + registry := newScriptedRegistry(bridge.Platform(fmt.Sprintf("reply-invariants-%d", seed)), text) + registry.setMediaSender(media) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + bodies := []string{"", " ", "hello", " padded ", "multi\nline"} + mimes := []string{"image/jpeg", "video/mp4", "audio/aac", "application/pdf"} + var cases []replyRefCase + + identityCount := 1 + random.Intn(3) + for index := 0; index < identityCount; index++ { + seedDispatchIdentity(t, store, fmt.Sprintf("identity-%d", index), fmt.Sprintf("+1555000%04d", index), clock.Now()) + } + incomingCount := 1 + random.Intn(4) + for index := 0; index < incomingCount; index++ { + message := sqlite.Message{ + MessageID: fmt.Sprintf("message-incoming-%d", index), + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: fmt.Sprintf("%040x", random.Uint64()), + Direction: sqlite.MessageDirectionIncoming, + Body: bodies[random.Intn(len(bodies))], + State: sqlite.MessageStateActive, + OccurredAtMS: 1_700_000_000_000 + random.Int63n(1_000_000_000), + } + author := "" + if random.Intn(4) != 0 { + identity := random.Intn(identityCount) + identityID := fmt.Sprintf("identity-%d", identity) + message.SenderIdentityID = &identityID + author = fmt.Sprintf("+1555000%04d", identity) + } + var attachmentMIMEs []string + for count := random.Intn(3); count > 0; count-- { + attachmentMIMEs = append(attachmentMIMEs, mimes[random.Intn(len(mimes))]) + } + projectReplyTarget(t, store, clock, message, attachmentMIMEs...) + mime := "" + if len(attachmentMIMEs) > 0 { + mime = attachmentMIMEs[0] + } + cases = append(cases, replyRefCase{ + name: message.MessageID, quotedID: message.MessageID, wantRemote: message.RemoteMessageID, + described: true, author: author, occurredMS: message.OccurredAtMS, body: message.Body, + mime: mime, hasMedia: len(attachmentMIMEs) > 0, + }) + } + + outgoingCount := 1 + random.Intn(4) + for index := 0; index < outgoingCount; index++ { + occurredMS := clock.Now().UnixMilli() + body := bodies[random.Intn(len(bodies))] + isMedia := random.Intn(3) == 0 + var submission Submission + mime := "" + if isMedia { + mime = mimes[random.Intn(len(mimes))] + submission = mustSendMedia(t, service, SendMediaCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("outgoing-%d", index)), + Content: bytes.NewReader([]byte(fmt.Sprintf("bytes-%d", index))), + Filename: "attachment.bin", + MIME: mime, + Caption: body, + }) + } else { + if strings.TrimSpace(body) == "" { + body = "outgoing text" + } + submission = mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("outgoing-%d", index)), + Body: body, + }) + } + row := mustOutboxItem(t, service, submission.OutboxID) + confirmed := random.Intn(3) != 0 + wantRemote := row.TransportRequestID + if confirmed { + wantRemote = fmt.Sprintf("%d", 1_700_000_000_000+int64(index)) + if isMedia { + media.mu.Lock() + media.steps = append(media.steps, sendStep{result: bridge.SendResult{RemoteMessageID: wantRemote}}) + media.mu.Unlock() + } else { + text.mu.Lock() + text.steps = append(text.steps, sendStep{result: bridge.SendResult{RemoteMessageID: wantRemote}}) + text.mu.Unlock() + } + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("seed %d: DispatchDue(outgoing %d) = %d, %v", seed, index, processed, err) + } + } else if _, err := service.Cancel(context.Background(), submission.OutboxID); err != nil { + t.Fatalf("seed %d: Cancel(outgoing %d): %v", seed, index, err) + } + cases = append(cases, replyRefCase{ + name: submission.LocalMessageID, quotedID: submission.LocalMessageID, wantRemote: wantRemote, + described: confirmed, occurredMS: occurredMS, body: body, mime: mime, hasMedia: isMedia, + }) + clock.Advance(time.Second) + } + + random.Shuffle(len(cases), func(i, j int) { cases[i], cases[j] = cases[j], cases[i] }) + for index, want := range cases { + // The first attempt fails before dispatch, so the reply is mapped twice. + text.mu.Lock() + text.steps = append(text.steps, + sendStep{err: bridge.OpError{ + Class: bridge.FailureTransient, Operation: "send_text", + Dispatch: bridge.DispatchNotCalled, Cause: errors.New("scripted retry"), + }}, + sendStep{result: bridge.SendResult{RemoteMessageID: fmt.Sprintf("reply-%d", index)}}, + ) + before := len(text.requests) + text.mu.Unlock() + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("reply-%d", index)), + Body: "reply", + ReplyToMessageID: want.quotedID, + }) + for attempt := 0; attempt < 2; attempt++ { + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("seed %d case %s attempt %d: DispatchDue = %d, %v", seed, want.name, attempt, processed, err) + } + clock.Advance(defaultRetryDelay + time.Second) + } + requests := text.snapshotRequests()[before:] + if len(requests) != 2 { + t.Fatalf("seed %d case %s: reply attempts = %d, want 2", seed, want.name, len(requests)) + } + if !messageRefEqual(requests[0].ReplyTo, requests[1].ReplyTo) { + t.Fatalf("seed %d case %s: retried ref %+v != first %+v", seed, want.name, requests[1].ReplyTo, requests[0].ReplyTo) + } + got := requests[1].ReplyTo + if got == nil || got.RemoteID != want.wantRemote { + t.Fatalf("seed %d case %s: ref = %+v, want remote %q", seed, want.name, got, want.wantRemote) + } + if got.SentAt.IsZero() == want.described { + t.Fatalf("seed %d case %s: ref = %+v, described = %v", seed, want.name, got, want.described) + } + if !want.described { + if *got != (bridge.MessageRef{RemoteID: want.wantRemote}) { + t.Fatalf("seed %d case %s: undescribed ref = %+v, want bare", seed, want.name, got) + } + continue + } + wantRef := bridge.MessageRef{ + RemoteID: want.wantRemote, AuthorID: want.author, SentAt: time.UnixMilli(want.occurredMS), + Text: want.body, HasAttachment: want.hasMedia, AttachmentMIME: want.mime, + } + if !messageRefEqual(got, &wantRef) { + t.Fatalf("seed %d case %s: ref = %+v, want %+v", seed, want.name, got, wantRef) + } + } +} + +// projectReplyTarget stores message as a projected (ingested) message with +// one pending attachment per MIME type, in ordinal order. +func projectReplyTarget( + t *testing.T, + store *sqlite.Store, + clock Clock, + message sqlite.Message, + attachmentMIMEs ...string, +) sqlite.Message { + t.Helper() + repository, err := sqlite.NewMessageRepository(store, clock.Now) + if err != nil { + t.Fatalf("NewMessageRepository(): %v", err) + } + inboxID := "inbox-" + message.MessageID + if _, err := repository.AppendInbox(context.Background(), sqlite.InboxRecord{ + InboxID: inboxID, + AccountID: message.AccountID, + Generation: 1, + DedupeKey: inboxID, + Codec: "reply-target-test", + CodecVersion: 1, + Payload: []byte("reply target"), + }); err != nil { + t.Fatalf("AppendInbox(%q): %v", inboxID, err) + } + attachments := make([]sqlite.MessageAttachment, 0, len(attachmentMIMEs)) + for ordinal, mime := range attachmentMIMEs { + size := int64(10 + ordinal) + attachments = append(attachments, sqlite.MessageAttachment{ + RemoteID: fmt.Sprintf("%s-attachment-%d", message.MessageID, ordinal), + Ordinal: int64(ordinal), + RemoteRef: []byte("{}"), + Filename: fmt.Sprintf("attachment-%d", ordinal), + MIME: mime, + SizeBytes: &size, + State: "pending", + }) + } + if err := repository.ProjectMessage(context.Background(), sqlite.MessageProjection{ + InboxID: inboxID, + Message: message, + Attachments: attachments, + }); err != nil { + t.Fatalf("ProjectMessage(%q): %v", message.MessageID, err) + } + projected, err := repository.GetMessage(context.Background(), message.MessageID) + if err != nil { + t.Fatalf("GetMessage(%q): %v", message.MessageID, err) + } + return projected +} + +func messageRefEqual(got, want *bridge.MessageRef) bool { + if got == nil || want == nil { + return got == want + } + return got.RemoteID == want.RemoteID && + got.AuthorID == want.AuthorID && + got.SentAt.Equal(want.SentAt) && + got.SentAt.IsZero() == want.SentAt.IsZero() && + got.Text == want.Text && + got.HasAttachment == want.HasAttachment && + got.AttachmentMIME == want.AttachmentMIME +} + +type failingAttachmentRepository struct { + messageRepository +} + +func (failingAttachmentRepository) FirstAttachmentMIME(context.Context, string) (string, bool, error) { + return "", false, errors.New("scripted attachment lookup failure") +} diff --git a/internal/messaging/types.go b/internal/messaging/types.go index a4b81aeb..bdd4a8ad 100644 --- a/internal/messaging/types.go +++ b/internal/messaging/types.go @@ -172,4 +172,9 @@ func (t systemTimer) C() <-chan time.Time { return t.Timer.C } // operations through its application-facing API. type messageRepository interface { GetMessage(context.Context, string) (sqlite.Message, error) + GetMessageByRemote( + ctx context.Context, + accountID, conversationID, remoteMessageID string, + ) (sqlite.Message, error) + FirstAttachmentMIME(ctx context.Context, messageID string) (string, bool, error) } diff --git a/internal/signallive/client.go b/internal/signallive/client.go index 5c58ddc5..e6ceedf6 100644 --- a/internal/signallive/client.go +++ b/internal/signallive/client.go @@ -1065,8 +1065,9 @@ func (b *Bridge) SendText(conversationID, body, replyToID string) (*db.Message, // SendTextRequest sends one text message through signal-cli's structured JSON // path and returns Signal's canonical outgoing identity. Signal uses the send // timestamp as its message ID; the durable request ID remains local dedupe -// metadata until reconciliation can make uncertain retries safe. -func (b *Bridge) SendTextRequest(conversationID, body, replyToID string) (timestampMS int64, err error) { +// metadata until reconciliation can make uncertain retries safe. A zero reply +// sends no quote. +func (b *Bridge) SendTextRequest(conversationID, body string, reply ReplyTarget) (timestampMS int64, err error) { body = strings.TrimSpace(body) if body == "" { return 0, errors.New("signal message body is required") @@ -1082,7 +1083,7 @@ func (b *Bridge) SendTextRequest(conversationID, body, replyToID string) (timest } args := []string{"--output=json", "-a", account, "send", "-m", body} - quoteArgs, err := b.signalQuoteArgs(replyToID, account) + quoteArgs, err := b.replyQuoteArgs(reply, account) if err != nil { return 0, err } @@ -1137,7 +1138,7 @@ func (b *Bridge) SendMedia(conversationID string, data []byte, filename, mime, c int64(len(data)), filename, caption, - replyToID, + ReplyTarget{RemoteID: replyToID}, true, ) if err != nil { @@ -1173,12 +1174,13 @@ func (b *Bridge) SendMedia(conversationID string, data []byte, filename, mime, c // SendMediaRequest streams one attachment to signal-cli and returns Signal's // canonical outgoing identity. Signal uses the send timestamp as its message // ID; the durable request ID remains local dedupe metadata until reconciliation -// can make uncertain retries safe. +// can make uncertain retries safe. A zero reply sends no quote. func (b *Bridge) SendMediaRequest( conversationID string, content io.Reader, size int64, - filename, mime, caption, replyToID string, + filename, mime, caption string, + reply ReplyTarget, ) (timestampMS int64, err error) { result, err := b.sendMediaRequest( conversationID, @@ -1186,7 +1188,7 @@ func (b *Bridge) SendMediaRequest( size, filename, caption, - replyToID, + reply, false, ) return result.timestampMS, err @@ -1206,7 +1208,8 @@ func (b *Bridge) sendMediaRequest( conversationID string, content io.Reader, size int64, - filename, caption, replyToID string, + filename, caption string, + reply ReplyTarget, retainOnSuccess bool, ) (result signalMediaRequestResult, err error) { account, err := b.usableAccount() @@ -1241,7 +1244,7 @@ func (b *Bridge) sendMediaRequest( // keeps the recipient available for parsing as the positional. // Reproduced with "No recipients given" when sending to an ACI-only contact. args = append(args, "--attachment="+attachmentPath) - quoteArgs, err := b.signalQuoteArgs(replyToID, account) + quoteArgs, err := b.replyQuoteArgs(reply, account) if err != nil { return result, err } @@ -4275,27 +4278,7 @@ func (b *Bridge) signalQuoteArgs(replyToID, account string) ([]string, error) { if target.TimestampMS == 0 { return nil, errors.New("signal reply target timestamp is unavailable") } - author := normalizeSignalAddress(target.SenderNumber) - if target.IsFromMe || addressesMatch(author, account) || author == "" { - author = account - } else { - author = b.resolveContactAddress(author) - } - if author == "" { - return nil, errors.New("signal reply target author is unavailable") - } - quoteBody := strings.TrimSpace(target.Body) - if quoteBody == "" && target.MediaID != "" { - quoteBody = signalAttachmentPlaceholder([]signalAttachment{{ContentType: target.MimeType}}) - } - if quoteBody == "" { - quoteBody = "Attachment" - } - return []string{ - "--quote-timestamp", strconv.FormatInt(target.TimestampMS, 10), - "--quote-author", author, - "--quote-message", quoteBody, - }, nil + return QuoteArgs(legacyReplyTarget(target), account, b.resolveContactAddress) } func (b *Bridge) writeLocalAttachment(data []byte, filename string) (string, error) { diff --git a/internal/signallive/client_test.go b/internal/signallive/client_test.go index a453816d..6088a0ad 100644 --- a/internal/signallive/client_test.go +++ b/internal/signallive/client_test.go @@ -141,7 +141,7 @@ func TestBridgeSendTextRequestRunsJSONSendAndReturnsSignalTimestamp(t *testing.T return success, nil } - timestamp, err := bridge.SendTextRequest(tc.conversationID, " hello from durable Signal ", "") + timestamp, err := bridge.SendTextRequest(tc.conversationID, " hello from durable Signal ", ReplyTarget{}) if err != nil { t.Fatalf("SendTextRequest(): %v", err) } @@ -202,7 +202,7 @@ func TestBridgeSendTextRequestIncludesSignalQuoteArguments(t *testing.T) { return success, nil } - if _, err := bridge.SendTextRequest(conversationID, "replying", "signal:reply-1"); err != nil { + if _, err := bridge.SendTextRequest(conversationID, "replying", ReplyTarget{RemoteID: "signal:reply-1"}); err != nil { t.Fatalf("SendTextRequest(): %v", err) } wantArgs := []string{ @@ -276,7 +276,7 @@ func TestBridgeSendTextRequestClassifiesStructuredAndOpaqueFailures(t *testing.T return tc.output, tc.commandErr } - _, err := bridge.SendTextRequest("signal:+15551234567", "hello", "") + _, err := bridge.SendTextRequest("signal:+15551234567", "hello", ReplyTarget{}) if err == nil || !IsCommandError(err) { t.Fatalf("SendTextRequest() error = %v (%T), want CommandError", err, err) } @@ -310,7 +310,7 @@ func TestBridgeSendTextRequestSucceedsOnMixedRecipientResults(t *testing.T) { return mixed, nil } - timestamp, err := bridge.SendTextRequest("signal-group:test-group", "hello group", "") + timestamp, err := bridge.SendTextRequest("signal-group:test-group", "hello group", ReplyTarget{}) if err != nil { t.Fatalf("SendTextRequest() on mixed results = %v, want success", err) } @@ -455,7 +455,7 @@ func TestBridgeSendMediaRequestStreamsAttachmentAndRemovesTempFileOnSuccess(t *t "photo.png", "image/png", "signal photo", - "", + ReplyTarget{}, ) if err != nil { t.Fatalf("SendMediaRequest(): %v", err) @@ -555,7 +555,7 @@ func TestBridgeSendMediaRequestClassifiesStructuredAndOpaqueFailures(t *testing. "photo.png", "image/png", "", - "", + ReplyTarget{}, ) if err == nil || !IsCommandError(err) { t.Fatalf("SendMediaRequest() error = %v (%T), want CommandError", err, err) @@ -615,7 +615,7 @@ func TestBridgeSendMediaRequestSucceedsOnMixedRecipientResults(t *testing.T) { "photo.png", "image/png", "group photo", - "", + ReplyTarget{}, ) if err != nil { t.Fatalf("SendMediaRequest() on mixed results = %v, want success", err) @@ -723,7 +723,7 @@ func TestBridgeSendMediaRequestRejectsOversizeReaderBeforeCommand(t *testing.T) "attachment.bin", "application/octet-stream", "", - "", + ReplyTarget{}, ) if err == nil || !strings.Contains(err.Error(), "exceeds declared size") { t.Fatalf("SendMediaRequest() error = %v, want declared-size error", err) diff --git a/internal/signallive/export_test.go b/internal/signallive/export_test.go index 030518bc..c80c3b87 100644 --- a/internal/signallive/export_test.go +++ b/internal/signallive/export_test.go @@ -2,6 +2,7 @@ package signallive import ( "bytes" + "context" "github.com/maxghenis/openmessage/internal/db" "github.com/rs/zerolog" @@ -50,3 +51,37 @@ func CaptureAndProcessReceiveLineForTest( processed, err := bridge.processReceiveLine(account, line, false) return captured, processed, err } + +// SetRunSignalCLIForTest replaces the signal-cli runner (contact and group +// refreshes included) for an external test and returns the restore function. +func SetRunSignalCLIForTest( + run func(ctx context.Context, configDir string, args ...string) ([]byte, error), +) func() { + original := runSignalCLI + runSignalCLI = run + return func() { runSignalCLI = original } +} + +// LegacyQuoteArgsForTest is the legacy-store quote lookup: what a reply quoting +// replyToID got before the v2 dispatcher described its target. +func LegacyQuoteArgsForTest( + store *db.Store, + contactByACI map[string]string, + account string, + replyToID string, +) ([]string, error) { + bridge := &Bridge{store: store, logger: zerolog.Nop(), contactByACI: contactByACI} + return bridge.signalQuoteArgs(replyToID, account) +} + +// ReplyQuoteArgsForTest is the quote SendTextRequest and SendMediaRequest +// build for reply. +func ReplyQuoteArgsForTest( + store *db.Store, + contactByACI map[string]string, + account string, + reply ReplyTarget, +) ([]string, error) { + bridge := &Bridge{store: store, logger: zerolog.Nop(), contactByACI: contactByACI} + return bridge.replyQuoteArgs(reply, account) +} diff --git a/internal/signallive/reply_quote.go b/internal/signallive/reply_quote.go new file mode 100644 index 00000000..da46f94a --- /dev/null +++ b/internal/signallive/reply_quote.go @@ -0,0 +1,146 @@ +package signallive + +import ( + "errors" + "strconv" + "strings" + "time" + + "github.com/maxghenis/openmessage/internal/db" +) + +// ReplyTarget is the message a durable Signal send quotes. RemoteID names it. +// The other fields describe the stored message; the v2 dispatcher fills them +// from the v2 store (bridge.MessageRef). A zero SentAt means they are absent, +// and the quote is resolved from the legacy store by RemoteID, as it always +// was (signalQuoteArgs). +type ReplyTarget struct { + RemoteID string + AuthorID string + SentAt time.Time + Text string + HasAttachment bool + AttachmentMIME string +} + +// QuoteArgs builds signal-cli's --quote-timestamp, --quote-author and +// --quote-message arguments for a reply to target sent from account. Signal +// identifies the quoted message by its author and sent timestamp. +// +// An empty AuthorID, or one matching account, quotes this account; any other +// author goes through resolveAuthor, which maps an ACI to its number when the +// contact is known. +// +// The timestamp is SentAt, the stored occurred time: the v2 decoder and the +// migration both set it to the Signal sent timestamp. One case differs. A +// message this account sent through the v2 outbox keeps its submit time as +// its occurred time, while confirming the send set its remote ID to the +// timestamp signal-cli reported. So for a message this account wrote, a +// RemoteID that is a decimal timestamp wins over SentAt. Every v2 writer gives +// such messages either that timestamp or a non-numeric ID ("local:" +// aliases, unconfirmed request IDs). Incoming messages carry SHA-1 IDs and +// always use SentAt. +// +// The quoted text is the trimmed body, else a placeholder naming the first +// attachment's kind, else "Attachment". +func QuoteArgs(target ReplyTarget, account string, resolveAuthor func(string) string) ([]string, error) { + author := normalizeSignalAddress(target.AuthorID) + self := author == "" || addressesMatch(author, account) + if self { + author = account + } else if resolveAuthor != nil { + author = resolveAuthor(author) + } + if author == "" { + return nil, errors.New("signal reply target author is unavailable") + } + + var timestampMS int64 + if remoteTimestamp, ok := signalTimestampID(target.RemoteID); ok && self { + timestampMS = remoteTimestamp + } else if !target.SentAt.IsZero() { + timestampMS = target.SentAt.UnixMilli() + } + if timestampMS <= 0 { + return nil, errors.New("signal reply target timestamp is unavailable") + } + + quoteBody := strings.TrimSpace(target.Text) + if quoteBody == "" && target.HasAttachment { + quoteBody = signalAttachmentPlaceholder([]signalAttachment{{ContentType: target.AttachmentMIME}}) + } + if quoteBody == "" { + quoteBody = "Attachment" + } + return []string{ + "--quote-timestamp", strconv.FormatInt(timestampMS, 10), + "--quote-author", author, + "--quote-message", quoteBody, + }, nil +} + +// replyQuoteArgs builds the quote for a durable send. A target the dispatcher +// described is quoted from that description and needs no legacy row; any +// other is looked up in the legacy store by RemoteID. +// +// One described target still quotes from its legacy row: a legacy message ID +// ("signal:" prefix) the legacy store holds. Of the v2 writers only the +// legacy-primary mirror (v2wire.MirrorReplyTarget) keys a message that way, +// and its copy carries no sender or attachment, so the legacy row stays the +// better source there, exactly as before. +func (b *Bridge) replyQuoteArgs(reply ReplyTarget, account string) ([]string, error) { + if reply.SentAt.IsZero() || b.holdsLegacyReplyTarget(reply.RemoteID) { + return b.signalQuoteArgs(reply.RemoteID, account) + } + return QuoteArgs(reply, account, b.resolveContactAddress) +} + +// holdsLegacyReplyTarget reports whether remoteID is a legacy Signal message +// ID whose row the legacy lookup can quote. +func (b *Bridge) holdsLegacyReplyTarget(remoteID string) bool { + remoteID = strings.TrimSpace(remoteID) + if b == nil || b.store == nil || !strings.HasPrefix(remoteID, "signal:") { + return false + } + target, err := b.store.GetMessageByID(remoteID) + return err == nil && target != nil && target.SourcePlatform == "signal" && target.TimestampMS != 0 +} + +// legacyReplyTarget describes a legacy Signal row for QuoteArgs exactly as the +// legacy lookup always read it: its own timestamp, and this account as author +// of a message it sent. RemoteID stays empty so the timestamp is the row's. +func legacyReplyTarget(target *db.Message) ReplyTarget { + authorID := target.SenderNumber + if target.IsFromMe { + authorID = "" + } + var sentAt time.Time + if target.TimestampMS != 0 { + sentAt = time.UnixMilli(target.TimestampMS) + } + return ReplyTarget{ + AuthorID: authorID, + SentAt: sentAt, + Text: target.Body, + HasAttachment: target.MediaID != "", + AttachmentMIME: target.MimeType, + } +} + +// signalTimestampID reports whether remoteID is a positive base-10 timestamp. +func signalTimestampID(remoteID string) (int64, bool) { + remoteID = strings.TrimSpace(remoteID) + if remoteID == "" { + return 0, false + } + for _, r := range remoteID { + if r < '0' || r > '9' { + return 0, false + } + } + timestamp, err := strconv.ParseInt(remoteID, 10, 64) + if err != nil || timestamp <= 0 { + return 0, false + } + return timestamp, true +} diff --git a/internal/signallive/reply_quote_differential_external_test.go b/internal/signallive/reply_quote_differential_external_test.go new file mode 100644 index 00000000..8be88ed9 --- /dev/null +++ b/internal/signallive/reply_quote_differential_external_test.go @@ -0,0 +1,398 @@ +package signallive_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/rand" + "path/filepath" + "slices" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/bridgeadapters/scripted" + signaladapter "github.com/maxghenis/openmessage/internal/bridgeadapters/signal" + "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/ingest" + "github.com/maxghenis/openmessage/internal/messaging" + "github.com/maxghenis/openmessage/internal/signallive" + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +const ( + quoteDiffAccountID = "signal-primary" + quoteDiffAccount = "+15551230000" + quoteDiffKnownACI = "9f4b50e3-ebf2-413c-a856-161756a6161a" + quoteDiffKnownE164 = "+15557654321" + quoteDiffOtherACI = "11111111-2222-3333-4444-555555555555" + quoteDiffGroupID = "cXVvdGUtZGlmZmVyZW50aWFsLWdyb3Vw" +) + +// The shapes of Signal traffic both receive paths store. Every kind must be +// generated at least once across the seeds (asserted below). +const ( + kindIncomingE164 = iota + kindIncomingKnownACI + kindIncomingUnknownACI + kindIncomingGroup + kindIncomingAttachmentOnly + kindIncomingTextAndAttachment + kindSyncSentText + kindSyncSentAttachmentOnly + quoteDiffKindCount +) + +type quoteDiffEntry struct { + kind int + line []byte + timestamp int64 + fromMe bool +} + +// TestV2DescribedQuoteMatchesLegacyLookup is the differential test for +// Signal quote-replies on a v2-primary daemon. Each seed feeds one random +// corpus of signal-cli receive lines through both retained paths: the legacy +// receive handler (writing the legacy store) and the durable tee, v2 decoder +// and ingest worker (writing the v2 store). For every message both stores +// then hold, it compares: +// +// - the legacy path: the quote signalQuoteArgs builds from the legacy row; +// - the new path: a reply submitted to the messaging service with the v2 +// message's ID, dispatched to a scripted Signal account, whose captured +// bridge.MessageRef the real adapter conversion hands to signallive's +// quote builder with an empty legacy store. +// +// The two must produce identical signal-cli arguments, and the new path must +// never need the legacy row (the ref is always described). +func TestV2DescribedQuoteMatchesLegacyLookup(t *testing.T) { + restore := signallive.SetRunSignalCLIForTest(func(context.Context, string, ...string) ([]byte, error) { + return nil, errors.New("signal-cli is not available in this test") + }) + t.Cleanup(restore) + + covered := make([]int, quoteDiffKindCount) + compared := 0 + for seed := int64(1); seed <= 12; seed++ { + seed := seed + t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { + compared += checkQuoteDifferential(t, seed, covered) + }) + } + for kind, count := range covered { + if count == 0 { + t.Errorf("corpus kind %d was never generated", kind) + } + } + t.Logf("compared %d messages; per-kind coverage %v", compared, covered) +} + +func checkQuoteDifferential(t *testing.T, seed int64, covered []int) int { + random := rand.New(rand.NewSource(seed)) + contacts := map[string]string{quoteDiffKnownACI: quoteDiffKnownE164} + dir := t.TempDir() + legacy := openQuoteDiffLegacy(t, filepath.Join(dir, "legacy.sqlite3")) + emptyLegacy := openQuoteDiffLegacy(t, filepath.Join(dir, "empty-legacy.sqlite3")) + v2Store, messages, sink, worker := openQuoteDiffV2(t, filepath.Join(dir, "v2.sqlite3")) + + entries := quoteDiffCorpus(random) + for index, entry := range entries { + covered[entry.kind]++ + captured, processed, err := signallive.CaptureAndProcessReceiveLineForTest( + legacy, t.TempDir(), contacts, quoteDiffAccount, entry.line, + ) + if err != nil || !processed { + t.Fatalf("entry %d legacy receive = (%v, %v), line %s", index, processed, err, entry.line) + } + record, ephemeral, err := ingest.BuildSignalIngress( + quoteDiffAccountID, 1, captured.Account, captured.Line, + captured.ResolvedSource, captured.ResolvedDestination, time.UnixMilli(entry.timestamp), + ) + if err != nil || record == nil || ephemeral != nil { + t.Fatalf("entry %d BuildSignalIngress = (%v, %v, %v)", index, record, ephemeral, err) + } + if err := sink.AppendIngress(context.Background(), *record); err != nil { + t.Fatalf("entry %d AppendIngress: %v", index, err) + } + } + drainQuoteDiffWorker(t, worker, messages) + + type key struct { + conversation string + timestamp int64 + fromMe bool + } + legacyRows := map[key]*db.Message{} + conversations, err := legacy.ListConversations(1000) + if err != nil { + t.Fatalf("legacy ListConversations: %v", err) + } + v2Rows := map[key]sqlite.Message{} + for _, conversation := range conversations { + rows, err := legacy.GetMessagesByConversation(conversation.ConversationID, 1000) + if err != nil { + t.Fatalf("legacy GetMessagesByConversation(%q): %v", conversation.ConversationID, err) + } + for _, row := range rows { + legacyRows[key{row.ConversationID, row.TimestampMS, row.IsFromMe}] = row + } + v2Conversation, err := v2Store.GetConversationByRemote(quoteDiffAccountID, conversation.ConversationID) + if err != nil { + t.Fatalf("v2 conversation for legacy %q: %v", conversation.ConversationID, err) + } + v2Messages, err := messages.ListMessagesByConversation(context.Background(), v2Conversation.ConversationID, 0, "", 1000) + if err != nil { + t.Fatalf("v2 ListMessagesByConversation(%q): %v", v2Conversation.ConversationID, err) + } + for _, message := range v2Messages { + v2Rows[key{conversation.ConversationID, message.OccurredAtMS, message.Direction == sqlite.MessageDirectionOutgoing}] = message + } + } + if len(legacyRows) != len(entries) || len(v2Rows) != len(entries) { + t.Fatalf("stored %d legacy and %d v2 messages for %d receive lines", len(legacyRows), len(v2Rows), len(entries)) + } + // occurred_at_ms is the Signal sent timestamp the decoder used, so the + // dispatcher's SentAt is a valid quote timestamp for incoming messages. + sentTimestamps := map[int64]bool{} + for _, entry := range entries { + sentTimestamps[entry.timestamp] = true + } + for rowKey, message := range v2Rows { + if !sentTimestamps[message.OccurredAtMS] { + t.Fatalf("v2 message %+v (key %+v) occurred at %d, not a sent timestamp of the corpus", message, rowKey, message.OccurredAtMS) + } + } + + registry := bridge.NewRegistry() + adapter := scripted.New(quoteDiffAccountID, bridge.PlatformSignal) + if err := registry.Register(adapter); err != nil { + t.Fatalf("Register(scripted Signal): %v", err) + } + service, err := messaging.NewMessageService(v2Store, registry, nil, messaging.SystemClock{}, messaging.CryptoIDSource{}) + if err != nil { + t.Fatalf("NewMessageService(): %v", err) + } + + compared := 0 + for rowKey, legacyRow := range legacyRows { + v2Row, ok := v2Rows[rowKey] + if !ok { + t.Fatalf("v2 store lacks the message legacy holds as %+v", legacyRow) + } + want, err := signallive.LegacyQuoteArgsForTest(legacy, contacts, quoteDiffAccount, legacyRow.MessageID) + if err != nil { + t.Fatalf("legacy quote for %+v: %v", legacyRow, err) + } + + adapter.EnqueueTextResult(bridge.SendResult{RemoteMessageID: fmt.Sprintf("%d", 1_800_000_000_000+int64(compared))}) + if _, err := service.SendText(context.Background(), messaging.SendTextCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: quoteDiffAccountID, + ConversationID: v2Row.ConversationID, + IdempotencyKey: fmt.Sprintf("quote-diff-%d-%d", seed, compared), + }, + Body: "reply", + ReplyToMessageID: v2Row.MessageID, + }); err != nil { + t.Fatalf("SendText(reply to %q): %v", v2Row.MessageID, err) + } + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reply to %q) = %d, %v", v2Row.MessageID, processed, err) + } + requests := adapter.TextRequests() + reply := requests[len(requests)-1].ReplyTo + if reply == nil || reply.SentAt.IsZero() { + t.Fatalf("reply to %q carried %+v, want a described ref", v2Row.MessageID, reply) + } + got, err := signallive.ReplyQuoteArgsForTest( + emptyLegacy, contacts, quoteDiffAccount, signaladapter.ReplyTarget(reply), + ) + if err != nil || !slices.Equal(got, want) { + t.Fatalf("seed %d: v2 quote for %+v = %q, %v; legacy row %+v quotes %q", + seed, v2Row, got, err, legacyRow, want) + } + compared++ + } + return compared +} + +func quoteDiffCorpus(random *rand.Rand) []quoteDiffEntry { + bodies := []string{"hi", " padded text ", "multi\nline", "emoji 🎉", "plain"} + mimes := []string{"image/jpeg", "image/png", "video/mp4", "audio/aac", "application/pdf", ""} + senders := []string{"+15551234567", "+15559876543"} + base := int64(1_700_000_000_000) + random.Int63n(1_000_000_000) + count := 6 + random.Intn(10) + entries := make([]quoteDiffEntry, 0, count) + for index := 0; index < count; index++ { + kind := random.Intn(quoteDiffKindCount) + // A minute apart, so the legacy sync-sent dedupe (15s drift) never + // folds two lines into one row. + timestamp := base + int64(index)*60_000 + random.Int63n(1_000) + body := bodies[random.Intn(len(bodies))] + attachment := map[string]any{ + "id": fmt.Sprintf("att-%d-%d", index, random.Intn(1_000_000)), + "contentType": mimes[random.Intn(len(mimes))], + "filename": "file.bin", + } + // Signal quotes by the sent timestamp, which signal-cli reports on the + // data (or sent) message. The envelope's own timestamp can differ, and + // a message without one falls back to the envelope's on both paths. + envelopeTimestamp := timestamp + if random.Intn(4) == 0 { + envelopeTimestamp = timestamp + 1 + random.Int63n(500) + } + messageTimestamp := map[string]any{"timestamp": timestamp} + if random.Intn(5) == 0 { + messageTimestamp = map[string]any{} + timestamp = envelopeTimestamp + } + envelope := map[string]any{"timestamp": envelopeTimestamp} + data := map[string]any{} + for field, value := range messageTimestamp { + data[field] = value + } + fromMe := false + switch kind { + case kindIncomingE164: + sender := senders[random.Intn(len(senders))] + envelope["source"], envelope["sourceNumber"], envelope["sourceUuid"] = sender, sender, quoteDiffOtherACI + envelope["sourceName"] = "E164 Friend" + data["message"] = body + case kindIncomingKnownACI: + envelope["sourceServiceId"], envelope["sourceName"] = quoteDiffKnownACI, "Known ACI" + data["message"] = body + case kindIncomingUnknownACI: + envelope["sourceServiceId"], envelope["sourceName"] = quoteDiffOtherACI, "Unknown ACI" + data["message"] = body + case kindIncomingGroup: + if random.Intn(2) == 0 { + envelope["sourceNumber"] = senders[random.Intn(len(senders))] + } else { + envelope["sourceServiceId"] = quoteDiffKnownACI + } + envelope["sourceName"] = "Group Member" + data["message"] = body + data["groupInfo"] = map[string]any{"groupId": quoteDiffGroupID, "type": "DELIVER"} + case kindIncomingAttachmentOnly: + envelope["sourceNumber"] = senders[random.Intn(len(senders))] + data["attachments"] = []any{attachment} + case kindIncomingTextAndAttachment: + envelope["sourceNumber"] = senders[random.Intn(len(senders))] + data["message"] = body + data["attachments"] = []any{attachment} + case kindSyncSentText, kindSyncSentAttachmentOnly: + fromMe = true + envelope["source"], envelope["sourceNumber"] = quoteDiffAccount, quoteDiffAccount + sent := map[string]any{} + for field, value := range messageTimestamp { + sent[field] = value + } + if kind == kindSyncSentText { + sent["message"] = body + } else { + sent["attachments"] = []any{attachment} + } + if random.Intn(3) == 0 { + sent["groupInfo"] = map[string]any{"groupId": quoteDiffGroupID, "type": "DELIVER"} + } else { + destination := senders[random.Intn(len(senders))] + sent["destination"], sent["destinationNumber"] = destination, destination + } + envelope["syncMessage"] = map[string]any{"sentMessage": sent} + } + if !fromMe { + envelope["dataMessage"] = data + } + line, err := json.Marshal(map[string]any{"account": quoteDiffAccount, "envelope": envelope}) + if err != nil { + panic(err) + } + entries = append(entries, quoteDiffEntry{kind: kind, line: line, timestamp: timestamp, fromMe: fromMe}) + } + return entries +} + +func openQuoteDiffLegacy(t *testing.T, path string) *db.Store { + t.Helper() + store, err := db.New(path) + if err != nil { + t.Fatalf("db.New(%q): %v", path, err) + } + t.Cleanup(func() { _ = store.Close() }) + return store +} + +func openQuoteDiffV2( + t *testing.T, + path string, +) (*sqlite.Store, *sqlite.MessageRepository, *ingest.Sink, *ingest.Worker) { + t.Helper() + store, err := sqlite.Open(path) + if err != nil { + t.Fatalf("sqlite.Open(): %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + nowMS := time.Now().UnixMilli() + if err := store.UpsertAccount(sqlite.Account{ + AccountID: quoteDiffAccountID, BridgeKey: "signal_cli", DisplayName: "Signal", + Mode: sqlite.AccountModeLive, Enabled: true, ConfigJSON: `{}`, + CreatedAtMS: nowMS, UpdatedAtMS: nowMS, + }); err != nil { + t.Fatalf("UpsertAccount(): %v", err) + } + messages, err := sqlite.NewMessageRepository(store, time.Now) + if err != nil { + t.Fatalf("NewMessageRepository(): %v", err) + } + reactions, err := sqlite.NewReactionRepository(store, time.Now) + if err != nil { + t.Fatalf("NewReactionRepository(): %v", err) + } + counters := &ingest.Counters{} + worker, err := ingest.NewWorker(ingest.WorkerConfig{ + Store: store, Messages: messages, Reactions: reactions, Counters: counters, + Decoders: []ingest.DecoderRegistration{{ + Codec: ingest.SignalJSONRPCCodec, + Platform: bridge.PlatformSignal, + Decoder: ingest.NewSignalDecoder(), + }}, + }) + if err != nil { + t.Fatalf("NewWorker(): %v", err) + } + nextID := 0 + sink, err := ingest.NewSink(ingest.SinkConfig{ + Messages: messages, Worker: worker, Counters: counters, + IDs: messaging.IDSourceFunc(func() (string, error) { + nextID++ + return fmt.Sprintf("quote-diff-inbox-%d", nextID), nil + }), + }) + if err != nil { + t.Fatalf("NewSink(): %v", err) + } + return store, messages, sink, worker +} + +func drainQuoteDiffWorker(t *testing.T, worker *ingest.Worker, messages *sqlite.MessageRepository) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- worker.Run(ctx) }() + defer func() { + cancel() + <-done + }() + deadline := time.Now().Add(10 * time.Second) + for { + pending, err := messages.Unprocessed(context.Background()) + if err == nil && len(pending) == 0 { + return + } + if time.Now().After(deadline) { + t.Fatalf("ingest worker left %d frames unprocessed (last error %v)", len(pending), err) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/internal/signallive/reply_quote_test.go b/internal/signallive/reply_quote_test.go new file mode 100644 index 00000000..051c88e5 --- /dev/null +++ b/internal/signallive/reply_quote_test.go @@ -0,0 +1,489 @@ +package signallive + +import ( + "context" + "errors" + "fmt" + "math/rand" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/db" + "github.com/rs/zerolog" +) + +const replyQuoteTestAccount = "+15551230000" + +func TestQuoteArgsRules(t *testing.T) { + const aci = "9f4b50e3-ebf2-413c-a856-161756a6161a" + resolve := func(value string) string { + if value == aci { + return "+15557654321" + } + return value + } + tests := []struct { + name string + target ReplyTarget + want []string + wantErr string + }{ + { + name: "own outbox send quotes the transport timestamp, not the submit time", + target: ReplyTarget{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(1_700_000_000_000), + Text: "sent through the outbox", + }, + want: quoteArgs("1700000000555", replyQuoteTestAccount, "sent through the outbox"), + }, + { + name: "own message under a migrated local alias quotes its occurred time", + target: ReplyTarget{ + RemoteID: "local:c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735", + SentAt: time.UnixMilli(1_700_000_000_321), + Text: "from the phone", + }, + want: quoteArgs("1700000000321", replyQuoteTestAccount, "from the phone"), + }, + { + name: "own message authored as the account address", + target: ReplyTarget{ + RemoteID: "1700000000999", + AuthorID: " " + replyQuoteTestAccount + " ", + SentAt: time.UnixMilli(1_700_000_000_111), + Text: "me", + }, + want: quoteArgs("1700000000999", replyQuoteTestAccount, "me"), + }, + { + name: "incoming message quotes its occurred time and author", + target: ReplyTarget{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_123), + Text: " hello there ", + }, + want: quoteArgs("1700000000123", "+15551234567", "hello there"), + }, + { + name: "incoming message never takes a numeric remote ID as its timestamp", + target: ReplyTarget{ + RemoteID: "1700000001000", + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_456), + Text: "legacy numeric id", + }, + want: quoteArgs("1700000000456", "+15551234567", "legacy numeric id"), + }, + { + name: "incoming ACI author resolves to its number", + target: ReplyTarget{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: aci, + SentAt: time.UnixMilli(1_700_000_000_123), + Text: "from an ACI", + }, + want: quoteArgs("1700000000123", "+15557654321", "from an ACI"), + }, + { + name: "body-less photo quotes a photo placeholder", + target: ReplyTarget{ + RemoteID: "1700000000555", SentAt: time.UnixMilli(1), HasAttachment: true, AttachmentMIME: "image/png", + }, + want: quoteArgs("1700000000555", replyQuoteTestAccount, "[Photo]"), + }, + { + name: "attachment of unknown type quotes the generic placeholder", + target: ReplyTarget{ + AuthorID: "+15551234567", SentAt: time.UnixMilli(1_700_000_000_123), Text: " ", HasAttachment: true, + }, + want: quoteArgs("1700000000123", "+15551234567", "[Attachment]"), + }, + { + name: "body-less message without an attachment", + target: ReplyTarget{AuthorID: "+15551234567", SentAt: time.UnixMilli(1_700_000_000_123)}, + want: quoteArgs("1700000000123", "+15551234567", "Attachment"), + }, + { + name: "no timestamp at all", + target: ReplyTarget{RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", AuthorID: "+15551234567"}, + wantErr: "signal reply target timestamp is unavailable", + }, + { + name: "own message whose request ID is not a timestamp and no occurred time", + target: ReplyTarget{RemoteID: "0f1e2d3c4b5a69788796a5b4c3d2e1f0"}, + wantErr: "signal reply target timestamp is unavailable", + }, + { + name: "pre-epoch occurred time", + target: ReplyTarget{AuthorID: "+15551234567", SentAt: time.UnixMilli(-5)}, + wantErr: "signal reply target timestamp is unavailable", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := QuoteArgs(test.target, replyQuoteTestAccount, resolve) + if test.wantErr != "" { + if err == nil || err.Error() != test.wantErr { + t.Fatalf("QuoteArgs() = %q, %v; want error %q", got, err, test.wantErr) + } + return + } + if err != nil || !slices.Equal(got, test.want) { + t.Fatalf("QuoteArgs() = %q, %v; want %q", got, err, test.want) + } + }) + } + + if _, err := QuoteArgs(ReplyTarget{SentAt: time.UnixMilli(1)}, "", resolve); err == nil || + err.Error() != "signal reply target author is unavailable" { + t.Fatalf("QuoteArgs() without an account = %v, want author unavailable", err) + } +} + +// TestQuoteArgsInvariants checks QuoteArgs over seeded random targets: +// +// - Shape: success yields exactly --quote-timestamp, --quote-author and +// --quote-message, with a positive decimal timestamp and non-empty author +// and text. +// - Timestamp source: a self-authored target whose RemoteID is a positive +// decimal uses it; every other target uses SentAt. QuoteArgs fails exactly +// when the chosen source is not positive. +// - Author: self-authored targets (empty AuthorID or the account) quote the +// account; others quote resolveAuthor(trimmed AuthorID). +// - Text: the trimmed body, else the attachment placeholder when the target +// has an attachment, else "Attachment". +// - Determinism: the same target always yields the same arguments. +func TestQuoteArgsInvariants(t *testing.T) { + random := rand.New(rand.NewSource(20261009)) + resolve := func(value string) string { return "resolved:" + value } + authors := []string{"", " ", replyQuoteTestAccount, " " + replyQuoteTestAccount, "+15551234567", "9f4b50e3-ebf2-413c-a856-161756a6161a"} + remoteIDs := []string{ + "", "0", "-17", "+17", "1700000000555", " 1700000000556 ", "99999999999999999999", + "f48818f15483f503bb133d92360ca8e2fbd8287e", "local:c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735", + "0f1e2d3c4b5a69788796a5b4c3d2e1f0", "17000000005x5", + } + bodies := []string{"", " ", "hi", " padded ", "multi\nline"} + mimes := []string{"", "image/png", "VIDEO/MP4", "audio/aac", "application/pdf"} + for iteration := 0; iteration < 5000; iteration++ { + target := ReplyTarget{ + RemoteID: remoteIDs[random.Intn(len(remoteIDs))], + AuthorID: authors[random.Intn(len(authors))], + Text: bodies[random.Intn(len(bodies))], + HasAttachment: random.Intn(2) == 0, + AttachmentMIME: mimes[random.Intn(len(mimes))], + } + switch random.Intn(4) { + case 0: + case 1: + target.SentAt = time.UnixMilli(-random.Int63n(1_000_000)) + default: + target.SentAt = time.UnixMilli(1 + random.Int63n(2_000_000_000_000)) + } + + got, err := QuoteArgs(target, replyQuoteTestAccount, resolve) + again, againErr := QuoteArgs(target, replyQuoteTestAccount, resolve) + if !slices.Equal(got, again) || (err == nil) != (againErr == nil) { + t.Fatalf("iteration %d: QuoteArgs(%+v) is not deterministic", iteration, target) + } + + author := strings.TrimSpace(target.AuthorID) + self := author == "" || author == replyQuoteTestAccount + wantAuthor := replyQuoteTestAccount + if !self { + wantAuthor = "resolved:" + author + } + var wantTimestamp int64 + if trimmed := strings.TrimSpace(target.RemoteID); self && trimmed != "" && + strings.Trim(trimmed, "0123456789") == "" { + if parsed, parseErr := strconv.ParseInt(trimmed, 10, 64); parseErr == nil && parsed > 0 { + wantTimestamp = parsed + } + } + if wantTimestamp == 0 && !target.SentAt.IsZero() { + wantTimestamp = target.SentAt.UnixMilli() + } + if wantTimestamp <= 0 { + if err == nil || err.Error() != "signal reply target timestamp is unavailable" { + t.Fatalf("iteration %d: QuoteArgs(%+v) = %q, %v; want unavailable timestamp", iteration, target, got, err) + } + continue + } + wantText := strings.TrimSpace(target.Text) + if wantText == "" && target.HasAttachment { + wantText = signalAttachmentPlaceholder([]signalAttachment{{ContentType: target.AttachmentMIME}}) + } + if wantText == "" { + wantText = "Attachment" + } + want := quoteArgs(strconv.FormatInt(wantTimestamp, 10), wantAuthor, wantText) + if err != nil || !slices.Equal(got, want) { + t.Fatalf("iteration %d: QuoteArgs(%+v) = %q, %v; want %q", iteration, target, got, err, want) + } + } +} + +// TestLegacyQuoteLookupMatchesPreChangeImplementation is a differential check +// of the legacy path: for random legacy Signal rows, signalQuoteArgs (now built +// on QuoteArgs) returns exactly what the pre-change implementation, kept below +// verbatim, returned. +func TestLegacyQuoteLookupMatchesPreChangeImplementation(t *testing.T) { + random := rand.New(rand.NewSource(7)) + store, err := db.New(filepath.Join(t.TempDir(), "messages.db")) + if err != nil { + t.Fatalf("db.New(): %v", err) + } + defer store.Close() + const aci = "9f4b50e3-ebf2-413c-a856-161756a6161a" + bridge := &Bridge{ + account: replyQuoteTestAccount, + connected: true, + configDir: t.TempDir(), + store: store, + logger: zerolog.Nop(), + contactByACI: map[string]string{aci: "+15557654321"}, + } + stubSignalCLIForQuoteTest(t) + + senders := []string{"", " ", replyQuoteTestAccount, "+15551234567", " +15551234567 ", aci, "11111111-2222-3333-4444-555555555555"} + bodies := []string{"", " ", "quoted", " padded quote "} + mimes := []string{"", "image/jpeg", "video/mp4", "audio/ogg", "application/zip"} + for index := 0; index < 400; index++ { + message := &db.Message{ + MessageID: fmt.Sprintf("signal:legacy-%d", index), + ConversationID: "signal:+15551234567", + SenderNumber: senders[random.Intn(len(senders))], + Body: bodies[random.Intn(len(bodies))], + TimestampMS: 1 + random.Int63n(2_000_000_000_000), + IsFromMe: random.Intn(3) == 0, + SourcePlatform: "signal", + } + if random.Intn(2) == 0 { + message.MediaID = fmt.Sprintf("signalatt:%d", index) + message.MimeType = mimes[random.Intn(len(mimes))] + } + if err := store.UpsertMessage(message); err != nil { + t.Fatalf("UpsertMessage(): %v", err) + } + got, err := bridge.signalQuoteArgs(message.MessageID, replyQuoteTestAccount) + want, wantErr := preChangeLegacyQuoteArgs(bridge, message, replyQuoteTestAccount) + if !slices.Equal(got, want) || errorText(err) != errorText(wantErr) { + t.Fatalf("row %+v: signalQuoteArgs = %q, %v; pre-change = %q, %v", message, got, err, want, wantErr) + } + } +} + +func TestReplyQuoteArgsUsesTheDescriptionWithoutALegacyRow(t *testing.T) { + store, err := db.New(filepath.Join(t.TempDir(), "messages.db")) + if err != nil { + t.Fatalf("db.New(): %v", err) + } + defer store.Close() + for _, legacy := range []*db.Store{store, nil} { + bridge := &Bridge{account: replyQuoteTestAccount, connected: true, store: legacy, logger: zerolog.Nop()} + got, err := bridge.replyQuoteArgs(ReplyTarget{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(1_700_000_000_000), + Text: "only v2 holds this", + }, replyQuoteTestAccount) + want := quoteArgs("1700000000555", replyQuoteTestAccount, "only v2 holds this") + if err != nil || !slices.Equal(got, want) { + t.Fatalf("replyQuoteArgs(legacy store %v) = %q, %v; want %q", legacy != nil, got, err, want) + } + } + + bridge := &Bridge{account: replyQuoteTestAccount, connected: true, store: store, logger: zerolog.Nop()} + if got, err := bridge.replyQuoteArgs(ReplyTarget{}, replyQuoteTestAccount); err != nil || got != nil { + t.Fatalf("replyQuoteArgs(no reply) = %q, %v; want no quote", got, err) + } + // An undescribed target still needs its legacy row, as before. + if _, err := bridge.replyQuoteArgs(ReplyTarget{RemoteID: "1700000000555"}, replyQuoteTestAccount); err == nil || + err.Error() != "signal reply target not found" { + t.Fatalf("replyQuoteArgs(undescribed, no legacy row) = %v, want not found", err) + } + if err := store.UpsertMessage(&db.Message{ + MessageID: "signal:legacy-quoted", + ConversationID: "signal:+15551234567", + SenderNumber: "+15551234567", + Body: "legacy row", + TimestampMS: 1_700_000_000_042, + SourcePlatform: "signal", + }); err != nil { + t.Fatalf("UpsertMessage(): %v", err) + } + got, err := bridge.replyQuoteArgs(ReplyTarget{RemoteID: "signal:legacy-quoted"}, replyQuoteTestAccount) + if want := quoteArgs("1700000000042", "+15551234567", "legacy row"); err != nil || !slices.Equal(got, want) { + t.Fatalf("replyQuoteArgs(legacy ID) = %q, %v; want %q", got, err, want) + } +} + +func TestReplyQuoteArgsQuotesALegacyIDFromItsLegacyRow(t *testing.T) { + store, err := db.New(filepath.Join(t.TempDir(), "messages.db")) + if err != nil { + t.Fatalf("db.New(): %v", err) + } + defer store.Close() + // The legacy-primary mirror keys its v2 copy of a reply target by the + // legacy ID and records neither sender nor attachment. + if err := store.UpsertMessage(&db.Message{ + MessageID: "signal:1700000000700", + ConversationID: "signal:+15551234567", + SenderNumber: "+15551234567", + TimestampMS: 1_700_000_000_700, + MediaID: "signalatt:mirrored", + MimeType: "image/png", + SourcePlatform: "signal", + }); err != nil { + t.Fatalf("UpsertMessage(): %v", err) + } + bridge := &Bridge{account: replyQuoteTestAccount, connected: true, store: store, logger: zerolog.Nop()} + mirrored := ReplyTarget{RemoteID: "signal:1700000000700", SentAt: time.UnixMilli(1_700_000_000_700)} + got, err := bridge.replyQuoteArgs(mirrored, replyQuoteTestAccount) + if want := quoteArgs("1700000000700", "+15551234567", "[Photo]"); err != nil || !slices.Equal(got, want) { + t.Fatalf("replyQuoteArgs(mirrored legacy ID) = %q, %v; want the legacy row's %q", got, err, want) + } + + // Without a legacy row the description is all there is. + mirrored.RemoteID = "signal:1700000000701" + mirrored.SentAt = time.UnixMilli(1_700_000_000_701) + got, err = bridge.replyQuoteArgs(mirrored, replyQuoteTestAccount) + if want := quoteArgs("1700000000701", replyQuoteTestAccount, "Attachment"); err != nil || !slices.Equal(got, want) { + t.Fatalf("replyQuoteArgs(legacy ID without a row) = %q, %v; want %q", got, err, want) + } +} + +func TestBridgeSendRequestsQuoteADescribedTargetTheLegacyStoreLacks(t *testing.T) { + store, err := db.New(filepath.Join(t.TempDir(), "messages.db")) + if err != nil { + t.Fatalf("db.New(): %v", err) + } + defer store.Close() + success, err := os.ReadFile(filepath.Join("testdata", "send-media-success.json")) + if err != nil { + t.Fatalf("read success fixture: %v", err) + } + const aci = "9f4b50e3-ebf2-413c-a856-161756a6161a" + bridge := &Bridge{ + account: replyQuoteTestAccount, + connected: true, + configDir: t.TempDir(), + store: store, + logger: zerolog.Nop(), + contactByACI: map[string]string{aci: "+15557654321"}, + } + originalRun := runSignalCLI + t.Cleanup(func() { runSignalCLI = originalRun }) + var captured []string + runSignalCLI = func(ctx context.Context, configDir string, args ...string) ([]byte, error) { + captured = append([]string{}, args...) + return success, nil + } + + ownSend := ReplyTarget{ + RemoteID: "1700000000555", + SentAt: time.UnixMilli(1_700_000_000_000), + Text: "sent through the v2 outbox", + } + if _, err := bridge.SendTextRequest("signal:+15551234567", "replying to myself", ownSend); err != nil { + t.Fatalf("SendTextRequest(own send): %v", err) + } + wantText := []string{ + "--output=json", "-a", replyQuoteTestAccount, "send", "-m", "replying to myself", + "--quote-timestamp", "1700000000555", + "--quote-author", replyQuoteTestAccount, + "--quote-message", "sent through the v2 outbox", + "+15551234567", + } + if !slices.Equal(captured, wantText) { + t.Fatalf("text args = %q, want %q", captured, wantText) + } + + incoming := ReplyTarget{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: aci, + SentAt: time.UnixMilli(1_700_000_000_123), + HasAttachment: true, + AttachmentMIME: "image/jpeg", + } + if _, err := bridge.SendMediaRequest( + "signal-group:group-id", + strings.NewReader("png-bytes"), + int64(len("png-bytes")), + "reply.png", + "image/png", + "", + incoming, + ); err != nil { + t.Fatalf("SendMediaRequest(incoming): %v", err) + } + if len(captured) < 2 || !strings.HasPrefix(captured[4], "--attachment=") { + t.Fatalf("media args = %q, want an attachment argument", captured) + } + wantMedia := []string{ + "--output=json", "-a", replyQuoteTestAccount, "send", captured[4], + "--quote-timestamp", "1700000000123", + "--quote-author", "+15557654321", + "--quote-message", "[Photo]", + "--group-id", "group-id", + } + if !slices.Equal(captured, wantMedia) { + t.Fatalf("media args = %q, want %q", captured, wantMedia) + } +} + +func quoteArgs(timestamp, author, message string) []string { + return []string{"--quote-timestamp", timestamp, "--quote-author", author, "--quote-message", message} +} + +func errorText(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +// stubSignalCLIForQuoteTest keeps resolveContactAddress's contact refresh +// from launching a real signal-cli. +func stubSignalCLIForQuoteTest(t *testing.T) { + t.Helper() + originalRun := runSignalCLI + t.Cleanup(func() { runSignalCLI = originalRun }) + runSignalCLI = func(context.Context, string, ...string) ([]byte, error) { + return nil, errors.New("signal-cli is not available in this test") + } +} + +// preChangeLegacyQuoteArgs is signalQuoteArgs's row-to-arguments logic as it +// stood before QuoteArgs, kept verbatim as the differential reference. +func preChangeLegacyQuoteArgs(b *Bridge, target *db.Message, account string) ([]string, error) { + if target.TimestampMS == 0 { + return nil, errors.New("signal reply target timestamp is unavailable") + } + author := normalizeSignalAddress(target.SenderNumber) + if target.IsFromMe || addressesMatch(author, account) || author == "" { + author = account + } else { + author = b.resolveContactAddress(author) + } + if author == "" { + return nil, errors.New("signal reply target author is unavailable") + } + quoteBody := strings.TrimSpace(target.Body) + if quoteBody == "" && target.MediaID != "" { + quoteBody = signalAttachmentPlaceholder([]signalAttachment{{ContentType: target.MimeType}}) + } + if quoteBody == "" { + quoteBody = "Attachment" + } + return []string{ + "--quote-timestamp", strconv.FormatInt(target.TimestampMS, 10), + "--quote-author", author, + "--quote-message", quoteBody, + }, nil +} diff --git a/internal/storage/sqlite/reply_target.go b/internal/storage/sqlite/reply_target.go new file mode 100644 index 00000000..cf64b02f --- /dev/null +++ b/internal/storage/sqlite/reply_target.go @@ -0,0 +1,48 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "fmt" +) + +// FirstAttachmentMIME reports the MIME type of a message's first attachment. +// It reads the lowest-ordinal message_attachments row, which ingested and +// migrated messages carry. An outgoing media message sent through the outbox +// has no such row; its attachment is the one its newest outbox intent was +// submitted with. ok is false when the message has neither. +func (r *MessageRepository) FirstAttachmentMIME( + ctx context.Context, + messageID string, +) (mime string, ok bool, err error) { + err = r.store.db.QueryRowContext(ctx, ` + SELECT mime + FROM message_attachments + WHERE message_id = ? + ORDER BY ordinal + LIMIT 1 + `, messageID).Scan(&mime) + if err == nil { + return mime, true, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return "", false, fmt.Errorf("first attachment of message %q: %w", messageID, err) + } + + err = r.store.db.QueryRowContext(ctx, ` + SELECT oa.mime + FROM outbox AS o + JOIN outbox_attachments AS oa ON oa.outbox_id = o.outbox_id + WHERE o.local_message_id = ? + ORDER BY o.created_at_ms DESC, o.outbox_id DESC, oa.ordinal + LIMIT 1 + `, messageID).Scan(&mime) + if errors.Is(err, sql.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("first outbox attachment of message %q: %w", messageID, err) + } + return mime, true, nil +} diff --git a/internal/storage/sqlite/reply_target_test.go b/internal/storage/sqlite/reply_target_test.go new file mode 100644 index 00000000..9ff24de2 --- /dev/null +++ b/internal/storage/sqlite/reply_target_test.go @@ -0,0 +1,107 @@ +package sqlite + +import ( + "context" + "path/filepath" + "testing" + "time" +) + +const replyTargetTestTimeMS = int64(1_700_000_000_000) + +func TestFirstAttachmentMIMEReadsStoredAttachmentsThenTheOutbox(t *testing.T) { + store, err := Open(filepath.Join(t.TempDir(), "store.sqlite3")) + if err != nil { + t.Fatalf("Open(): %v", err) + } + t.Cleanup(func() { + if err := store.Close(); err != nil { + t.Errorf("Close(): %v", err) + } + }) + now := func() time.Time { return time.UnixMilli(replyTargetTestTimeMS) } + messages, err := NewMessageRepository(store, now) + if err != nil { + t.Fatalf("NewMessageRepository(): %v", err) + } + outbox, err := NewOutboxRepository(store, now) + if err != nil { + t.Fatalf("NewOutboxRepository(): %v", err) + } + seedMessageAccount(t, store, "account-a", "test") + seedMessageIdentity(t, store, "identity-a", "account-a") + seedMessageConversation(t, store, "conversation-a", "account-a") + + insertMessage := func(messageID string) { + t.Helper() + mustExec(t, store.db, ` + INSERT INTO messages ( + message_id, conversation_id, account_id, remote_message_id, + sender_identity_id, direction, body, reply_to_remote_id, state, + occurred_at_ms, created_at_ms, updated_at_ms + ) VALUES (?, 'conversation-a', 'account-a', ?, NULL, 'incoming', '', NULL, + 'active', ?, ?, ?) + `, messageID, "remote-"+messageID, replyTargetTestTimeMS, replyTargetTestTimeMS, replyTargetTestTimeMS) + } + insertAttachment := func(messageID string, ordinal int, mime string) { + t.Helper() + mustExec(t, store.db, ` + INSERT INTO message_attachments ( + message_id, ordinal, mime, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?) + `, messageID, ordinal, mime, replyTargetTestTimeMS, replyTargetTestTimeMS) + } + + insertMessage("message-ordinals") + insertAttachment("message-ordinals", 2, "video/mp4") + insertAttachment("message-ordinals", 1, "image/jpeg") + insertMessage("message-none") + + media := outboxTestMediaItem("reply-target-media") + if _, _, err := outbox.EnqueueOutgoingMediaMessage( + context.Background(), + media, + outboxTestOutgoingMessage(media, ""), + outboxTestAttachment(), + ); err != nil { + t.Fatalf("EnqueueOutgoingMediaMessage(): %v", err) + } + text := outboxTestItem("reply-target-text") + if _, _, err := outbox.EnqueueOutgoingMessage( + context.Background(), + text, + outboxTestOutgoingMessage(text, "text only"), + ); err != nil { + t.Fatalf("EnqueueOutgoingMessage(): %v", err) + } + // A stored attachment row outranks the outbox's for the same message. + shadowed := outboxTestMediaItem("reply-target-shadowed") + if _, _, err := outbox.EnqueueOutgoingMediaMessage( + context.Background(), + shadowed, + outboxTestOutgoingMessage(shadowed, ""), + outboxTestAttachment(), + ); err != nil { + t.Fatalf("EnqueueOutgoingMediaMessage(shadowed): %v", err) + } + insertAttachment(shadowed.LocalMessageID, 0, "audio/aac") + + for _, test := range []struct { + messageID string + wantMIME string + wantOK bool + }{ + {messageID: "message-ordinals", wantMIME: "image/jpeg", wantOK: true}, + {messageID: "message-none"}, + {messageID: media.LocalMessageID, wantMIME: "image/png", wantOK: true}, + {messageID: text.LocalMessageID}, + {messageID: shadowed.LocalMessageID, wantMIME: "audio/aac", wantOK: true}, + {messageID: "message-missing"}, + } { + mime, ok, err := messages.FirstAttachmentMIME(context.Background(), test.messageID) + if err != nil || mime != test.wantMIME || ok != test.wantOK { + t.Errorf("FirstAttachmentMIME(%q) = (%q, %v, %v), want (%q, %v, nil)", + test.messageID, mime, ok, err, test.wantMIME, test.wantOK) + } + } +}