diff --git a/cmd/r5_signal_reaction_test.go b/cmd/r5_signal_reaction_test.go new file mode 100644 index 00000000..a7aa3a7e --- /dev/null +++ b/cmd/r5_signal_reaction_test.go @@ -0,0 +1,254 @@ +package cmd + +import ( + "context" + "fmt" + "path/filepath" + "slices" + "strconv" + "sync" + "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" +) + +// r5ReactingSignal gives the scripted Signal account, which has no reaction +// sender of its own, one that records each request and confirms it. +type r5ReactingSignal struct { + *scripted.Adapter + + mu sync.Mutex + requests []bridge.ReactionRequest +} + +func (a *r5ReactingSignal) SendReaction(_ context.Context, req bridge.ReactionRequest) (bridge.SendResult, error) { + a.mu.Lock() + a.requests = append(a.requests, req) + a.mu.Unlock() + return bridge.SendResult{}, nil +} + +func (a *r5ReactingSignal) reactionRequests() []bridge.ReactionRequest { + a.mu.Lock() + defer a.mu.Unlock() + return slices.Clone(a.requests) +} + +// TestR5SignalReactionsTargetTheSentTimestampOnV2Primary runs Signal reactions +// end to end on the quote-reply test's harness: a migrated v2-primary stack +// with a scripted Signal account. A reaction goes through the messaging +// service (no HTTP route submits v2 reactions; /api/react still calls the +// legacy bridge), the durable dispatcher and the bridge request, and the +// captured target then goes through the real Signal adapter conversion into +// signallive's argument builder. It covers a live-ingested incoming message +// (SHA-1 remote ID), this account's own send through the v2 outbox, and the +// migrated messages whose Signal identity v2 cannot vouch for, which must be +// refused: an incoming row under a decimal ID, an own "local:" row, and a +// scheduled send caught in "sending". +func TestR5SignalReactionsTargetTheSentTimestampOnV2Primary(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 := &r5ReactingSignal{Adapter: scripted.New(r5SignalAccountID, bridge.PlatformSignal)} + if err := stack.RegisterAdapter(signal); err != nil { + t.Fatalf("register scripted Signal adapter: %v", err) + } + 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) + signalConversation := v2keys.DeriveID("conversation", r5SignalAccountID, r5SignalConversation) + + // react submits one reaction, waits for the dispatcher to hand it to the + // (scripted) transport, and returns the target the bridge request carried + // with what the Signal transport makes of it: the signal-cli target + // arguments, or the error it stops on before running signal-cli. + react := func(key, targetMessageID string) (bridge.MessageRef, []string, error) { + t.Helper() + before := len(signal.reactionRequests()) + submission, err := stack.Service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: r5SignalAccountID, ConversationID: signalConversation, IdempotencyKey: key, + }, + TargetMessageID: targetMessageID, + Emoji: "👍", + }) + if err != nil { + t.Fatalf("SendReaction(%q): %v", targetMessageID, err) + } + waitR5(t, "reaction "+key, func() bool { + delivery, getErr := stack.Service.Get(context.Background(), submission.OutboxID) + return getErr == nil && delivery.State == messaging.OutboxConfirmed + }) + requests := signal.reactionRequests() + if len(requests) != before+1 { + t.Fatalf("reaction %q reached the transport %d times, want once", key, len(requests)-before) + } + request := requests[before] + if request.Conversation.RemoteID != r5SignalConversation { + t.Fatalf("reaction %q conversation = %q, want %q", key, request.Conversation.RemoteID, r5SignalConversation) + } + args, err := signallive.ReactionTargetArgs(signaladapter.ReactionTarget(request.Target), r5SignalAccountAddress) + return request.Target, args, err + } + targetArgs := func(author string, timestamp int64) []string { + return []string{"-a", author, "-t", strconv.FormatInt(timestamp, 10)} + } + + t.Run("reaction to a live-ingested incoming message", func(t *testing.T) { + const body = "r5 live incoming to react to" + timestamp := fixture.BaseMS + 50_000 + line := []byte(fmt.Sprintf( + `{"account":%q,"envelope":{"sourceServiceId":%q,"sourceName":"R5 Signal Reacted","timestamp":%d,"dataMessage":{"timestamp":%d,"message":%q}}}`, + r5SignalAccountAddress, r5SignalACI, timestamp+9, timestamp, body, + )) + record, ephemeral, err := ingest.BuildSignalIngress( + r5SignalAccountID, 1, r5SignalAccountAddress, line, "", "", time.UnixMilli(timestamp+9), + ) + 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 + }) + + target, args, err := react("r5-react-incoming", incoming.MessageID) + // The v2 remote ID is a SHA-1, which signal-cli's -t cannot take. + if want := v2keys.SignalIncomingSourceID(r5SignalConversation, r5SignalACI, timestamp); target.RemoteID != want { + t.Fatalf("reaction target remote ID = %q, want the SHA-1 %q", target.RemoteID, want) + } + if want := targetArgs(r5SignalACI, timestamp); err != nil || !slices.Equal(args, want) { + t.Fatalf("reaction target = %q, %v; want %q", args, err, want) + } + }) + + t.Run("reaction to a migrated incoming message under a decimal ID is refused", func(t *testing.T) { + // The fixture's incoming Signal row has a numeric legacy ID that is + // not its timestamp. No Signal receiver keys a message that way, so + // nothing vouches that its stored time is the one Signal knows it by: + // neither the ID nor the time may reach -t. + legacyFixture, err := db.New(fixture.StorePath) + if err != nil { + t.Fatalf("open legacy fixture store: %v", err) + } + defer legacyFixture.Close() + row, err := legacyFixture.GetMessageByID("signal:1700000001000") + if err != nil || row == nil || row.IsFromMe || row.TimestampMS == 1700000001000 { + t.Fatalf("legacy fixture row = %+v, %v; want an incoming row whose ID is not its timestamp", row, err) + } + v2ID := v2keys.DeriveID("message", r5SignalAccountID, r5SignalConversation+"\x1f1700000001000") + target, args, err := react("r5-react-migrated-incoming", v2ID) + if target.RemoteID != "1700000001000" || target.Outgoing || target.AuthorID != row.SenderNumber || + !target.SentAt.Equal(time.UnixMilli(row.TimestampMS)) { + t.Fatalf("reaction target = %+v, want the migrated incoming row as stored", target) + } + if err == nil || err.Error() != "signal reaction target timestamp is unavailable" || args != nil { + t.Fatalf("reaction target = %q, %v; want no timestamp to send", args, err) + } + }) + + t.Run("reactions to migrated own messages with no known Signal timestamp are refused", func(t *testing.T) { + // A "local:" row keeps whatever time the legacy store held, which for + // a send the legacy SendText made is the wall clock after signal-cli + // returned. A scheduled send caught in "sending" is imported under a + // derived request ID with its creation time and no outbox row. + sendingRequestID := v2keys.DeriveID("transport_request", r5SignalAccountID, "r5-sending") + for _, migrated := range []struct{ name, remoteID string }{ + {name: "local alias", remoteID: "local:abc123r5"}, + {name: "scheduled send left in sending", remoteID: sendingRequestID}, + } { + v2ID := v2keys.DeriveID("message", r5SignalAccountID, r5SignalConversation+"\x1f"+migrated.remoteID) + stored, err := messages.GetMessage(context.Background(), v2ID) + if err != nil || stored.Direction != sqlite.MessageDirectionOutgoing || stored.OccurredAtMS <= 0 { + t.Fatalf("migrated %s = %+v, %v; want an own message with a stored time", migrated.name, stored, err) + } + target, args, err := react("r5-react-migrated-own-"+migrated.name, v2ID) + if target.RemoteID != migrated.remoteID || !target.Outgoing || + !target.SentAt.Equal(time.UnixMilli(stored.OccurredAtMS)) { + t.Fatalf("reaction target for migrated %s = %+v, want the own message as stored", migrated.name, target) + } + if err == nil || err.Error() != "signal reaction target timestamp is unavailable" || args != nil { + t.Fatalf("reaction target for migrated %s = %q, %v; want no timestamp to send", migrated.name, args, err) + } + } + }) + + t.Run("reaction to this account's own v2 outbox send", func(t *testing.T) { + signal.EnqueueTextResult(bridge.SendResult{RemoteMessageID: "1752800000030"}) + own, err := stack.Service.SendText(context.Background(), messaging.SendTextCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: r5SignalAccountID, ConversationID: signalConversation, IdempotencyKey: "r5-react-own-send", + }, + Body: "r5 own outbox send to react to", + }) + if err != nil { + t.Fatalf("submit own send: %v", err) + } + waitR5(t, "own send", func() bool { + delivery, getErr := stack.Service.Get(context.Background(), own.OutboxID) + return getErr == nil && delivery.State == messaging.OutboxConfirmed && delivery.RemoteMessageID == "1752800000030" + }) + stored, err := messages.GetMessage(context.Background(), own.LocalMessageID) + if err != nil { + t.Fatalf("GetMessage(own send): %v", err) + } + + target, args, err := react("r5-react-own", own.LocalMessageID) + // The stored occurred time is the submit time; the reaction targets + // the timestamp signal-cli reported for the send. + if !target.Outgoing || !target.SentAt.Equal(time.UnixMilli(stored.OccurredAtMS)) || + stored.OccurredAtMS == 1752800000030 { + t.Fatalf("own send target = %+v, stored occurred at %d; want the submit time", target, stored.OccurredAtMS) + } + if want := targetArgs(r5SignalAccountAddress, 1752800000030); err != nil || !slices.Equal(args, want) { + t.Fatalf("reaction target = %q, %v; want %q", args, err, want) + } + }) +} diff --git a/docs/agent-runbook.md b/docs/agent-runbook.md index fcd3c4d0..8da4b444 100644 --- a/docs/agent-runbook.md +++ b/docs/agent-runbook.md @@ -1504,6 +1504,83 @@ Rules worth knowing when a quote looks wrong: 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. +### Signal reactions through the v2 outbox + +Signal names the message a reaction targets by its author and sent timestamp +(`signal-cli sendReaction -a -t `), never by an ID +OpenMessage stores. The dispatcher (`messaging.targetRefForLease`) puts the +target's sender, direction (`Outgoing`) and `occurred_at_ms` on the request's +`bridge.MessageRef`, and `signallive.ReactionTargetArgs` sends a reaction only +when the store vouches for both halves of that name: + +- **An incoming message** is named by its stored sender and `occurred_at_ms`, + and only when its remote ID is a SHA-1 (40 hex digits, + `v2keys.SignalIncomingSourceID`). That is the ID a Signal receiver gives a + message it keyed by sender and sent timestamp, and each writer that uses it + stores that timestamp as the occurred time: the v2 decoder, the legacy + receiver, and a Signal Desktop import of a row that has a sent time. The ID + itself is not sent. +- **A message this account sent** is named by this account and its remote ID, + and only when that ID is a decimal timestamp. That is how a send Signal + accepted is stored when its timestamp was kept: an outbox confirm sets the ID + to the timestamp signal-cli reported, a sync message from the phone arrives + under its own, and the legacy `SendMedia` and legacy-primary projector keyed + their rows by signal-cli's. Its `occurred_at_ms` is not used. This is stricter than the + quote rule above, because a quote carries the quoted text with it and a + reaction carries nothing but the name. + +Anything else stops before signal-cli runs. The row goes `not_dispatched` +(`error_class: transient`, `error_code: send_reaction` on +`GET /api/v1/outbox/`) and retries every 5 s with no cap. The reason is in +the `error_detail` column of the `outbox` table in the v2 store: + +- `signal reaction target timestamp is unavailable`, for this account's own + message: its remote ID is not a Signal timestamp. + - *Still on its outbox request ID.* A pending send: the reaction goes out on + a retry once the send is confirmed. A canceled or failed send: never. An + uncertain one: not while it is unresolved. Also a send Signal accepted + whose local confirm failed (`store_failed`): the outbox row holds the + timestamp as its result, but the message is not moved to it until that + confirm is repaired (`RepairStoreFailed`). + - *A migrated `local:` message.* The legacy `SendText` stamped its row + with the wall clock after signal-cli returned, not with Signal's + timestamp. The phone's own sends were stored under the same alias with the + right time, and nothing tells the two apart, so none of them can be + reacted to through the outbox. + - *A migrated scheduled send caught in `sending`*: imported under a derived + request ID with its creation time and no outbox row. +- `signal reaction target timestamp is unavailable`, for an incoming message: + its remote ID is not a SHA-1, or it has no occurred time. + - *`received:`*: a Signal Desktop row that had no sent time. The + importer stores it under the time it was received and marks its source ID + (`v2keys.SignalReceivedSourceID`). + - *`signal:`*: a legacy ID the legacy-primary mirror kept. +- `signal reaction target author is unavailable`: an incoming message stored + with no sender. The legacy receiver stored a group message with no source + that way, and the legacy-primary mirror (`v2wire.MirrorReplyTarget`) records + none. + +A reaction that can never be named keeps retrying until it is canceled +(`POST /api/v1/outbox//cancel`). + +One gap remains that nothing can mark. Before the importer marked them, a +Signal Desktop row with no sent time was stored under a plain SHA-1 of its +received time. A re-import rewrites that source ID in the legacy store, but a +v2 store migrated before then still holds the row under the unmarked hash, +indistinguishable from a message keyed by its sent timestamp. A reaction to +it names the received time. + +Before this the transport passed the v2 remote ID to `-t`. For a decoded +incoming message that is a SHA-1. signal-cli 0.14.8 refuses a non-integer +while parsing its arguments (`could not convert '' to integer (64 bits)`, +exit 1), and the row ended `uncertain` with nothing sent. + +No surface submits a reaction to the v2 outbox yet. The web UI's `/api/react` +and the `react_to_message` MCP tool both call the legacy +`signallive.Bridge.SendReaction`, which looks the target up in the legacy +`messages.db` by message ID. Only `MessageService.SendReaction` (tests, so +far) reaches the path above. + ## 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 6a85eb05..80bd262e 100644 --- a/internal/bridge/contracts.go +++ b/internal/bridge/contracts.go @@ -64,15 +64,20 @@ type ConversationRef struct { // 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. +// when the stored message names no sender. That is always the case for a +// message this account sent, but an incoming message can lack a sender too +// (the legacy Signal receiver stored group messages with no source, and the +// legacy-primary mirror records none), so an empty AuthorID does not by itself +// mean this account. Outgoing reports that this account sent the message. +// SentAt is the message's occurred time. Reaction and read refs always +// describe the message they name. A reply ref with a zero SentAt is +// undescribed (the dispatcher holds no message under RemoteID, or the quoted +// message is an outgoing one still on its outbox request ID), so only its +// RemoteID is meaningful. type MessageRef struct { RemoteID string AuthorID string + Outgoing bool SentAt time.Time // Text, HasAttachment and AttachmentMIME are filled for reply targets // only. Text is the stored body. HasAttachment reports whether the message diff --git a/internal/bridgeadapters/signal/reaction_target_test.go b/internal/bridgeadapters/signal/reaction_target_test.go new file mode 100644 index 00000000..4cb7da05 --- /dev/null +++ b/internal/bridgeadapters/signal/reaction_target_test.go @@ -0,0 +1,154 @@ +package signal + +import ( + "context" + "errors" + "reflect" + "slices" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/signallive" +) + +func TestReactionTargetCarriesTheFieldsThatNameAMessage(t *testing.T) { + ref := bridge.MessageRef{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: "+15551234567", + Outgoing: true, + SentAt: time.UnixMilli(1_700_000_000_123), + Text: "reply-only field", + HasAttachment: true, + AttachmentMIME: "image/jpeg", + } + got := reflect.ValueOf(ReactionTarget(ref)) + source := reflect.ValueOf(ref) + // Every ReactionTarget field must be the MessageRef field of the same + // name, so a field added to it cannot be left unfilled here silently. + var names []string + for index := 0; index < got.NumField(); index++ { + name := got.Type().Field(index).Name + names = append(names, name) + field := source.FieldByName(name) + if !field.IsValid() { + t.Fatalf("bridge.MessageRef has no %s field for signallive.ReactionTarget", name) + } + if !reflect.DeepEqual(got.Field(index).Interface(), field.Interface()) { + t.Fatalf("ReactionTarget().%s = %v, want %v", name, got.Field(index).Interface(), field.Interface()) + } + if field.IsZero() { + t.Fatalf("fixture leaves MessageRef.%s zero; set it so the copy is checked", name) + } + } + // Signal names a reaction's target by author and sent timestamp. Outgoing + // says whose message it is, and RemoteID is the timestamp of this + // account's own sends. + if want := []string{"RemoteID", "AuthorID", "Outgoing", "SentAt"}; !slices.Equal(names, want) { + t.Fatalf("signallive.ReactionTarget fields = %v, want %v", names, want) + } +} + +func TestSendReactionHandsTheDispatchersDescriptionToThePoller(t *testing.T) { + tests := []struct { + name string + target bridge.MessageRef + }{ + { + name: "incoming message", + target: bridge.MessageRef{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: "9f4b50e3-ebf2-413c-a856-161756a6161a", + SentAt: time.UnixMilli(1_700_000_000_123), + }, + }, + { + name: "this account's confirmed send", + target: bridge.MessageRef{RemoteID: "1700000000555", Outgoing: true, SentAt: time.UnixMilli(1_700_000_000_000)}, + }, + { + name: "this account's send still on its request ID", + target: bridge.MessageRef{ + RemoteID: "0f1e2d3c4b5a69788796a5b4c3d2e1f0", Outgoing: true, SentAt: time.UnixMilli(1_700_000_000_000), + }, + }, + { + name: "incoming message with no stored sender", + target: bridge.MessageRef{RemoteID: "1700000000123", SentAt: time.UnixMilli(1_700_000_000_123)}, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + poller := newFakePoller() + poller.status = signallive.StatusSnapshot{Connected: true, Paired: true, Account: "+15551230000"} + adapter := &Adapter{accountID: "signal-primary", poller: poller} + if _, err := adapter.SendReaction(context.Background(), bridge.ReactionRequest{ + AccountID: "signal-primary", + Conversation: bridge.ConversationRef{RemoteID: "signal:+15551234567"}, + Target: tc.target, + Emoji: "👍", + Action: bridge.ReactionAdd, + }); err != nil { + t.Fatalf("SendReaction(): %v", err) + } + want := signallive.ReactionTarget{ + RemoteID: tc.target.RemoteID, AuthorID: tc.target.AuthorID, + Outgoing: tc.target.Outgoing, SentAt: tc.target.SentAt, + } + if got := poller.lastReactionRequest().target; got != want { + t.Fatalf("reaction target handed to signallive = %+v, want %+v", got, want) + } + }) + } +} + +// A target signallive cannot name fails before signal-cli runs: this account's +// send that has no Signal timestamp yet, or an incoming message with no stored +// sender. The adapter must report that as not dispatched, so the outbox +// retries instead of marking the reaction uncertain. +func TestSendReactionReportsATargetItCannotNameAsNotDispatched(t *testing.T) { + tests := []struct { + name string + target bridge.MessageRef + }{ + { + name: "this account's send still on its request ID", + target: bridge.MessageRef{ + RemoteID: "0f1e2d3c4b5a69788796a5b4c3d2e1f0", Outgoing: true, SentAt: time.UnixMilli(1_700_000_000_000), + }, + }, + { + name: "incoming message with no stored sender", + target: bridge.MessageRef{RemoteID: "1700000000123", SentAt: time.UnixMilli(1_700_000_000_123)}, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, targetErr := signallive.ReactionTargetArgs(ReactionTarget(tc.target), "+15551230000") + if targetErr == nil { + t.Fatal("ReactionTargetArgs() succeeded, want a target it cannot name") + } + poller := newFakePoller() + poller.status = signallive.StatusSnapshot{Connected: true, Paired: true, Account: "+15551230000"} + poller.reactionErr = targetErr + adapter := &Adapter{accountID: "signal-primary", poller: poller} + + _, err := adapter.SendReaction(context.Background(), bridge.ReactionRequest{ + AccountID: "signal-primary", + Conversation: bridge.ConversationRef{RemoteID: "signal:+15551234567"}, + Target: tc.target, + Emoji: "👍", + }) + var operationError bridge.OpError + if !errors.As(err, &operationError) || + operationError.Class != bridge.FailureTransient || + operationError.Dispatch != bridge.DispatchNotCalled || + !errors.Is(operationError.Cause, targetErr) { + t.Fatalf("SendReaction() error = %+v, want a retryable not-dispatched failure caused by %v", err, targetErr) + } + if got := poller.appliedCount(); got != 0 { + t.Fatalf("a target it cannot name applied %d lifecycle transitions, want 0", got) + } + }) + } +} diff --git a/internal/bridgeadapters/signal/reply_target_test.go b/internal/bridgeadapters/signal/reply_target_test.go index 925757d1..11be88e0 100644 --- a/internal/bridgeadapters/signal/reply_target_test.go +++ b/internal/bridgeadapters/signal/reply_target_test.go @@ -19,6 +19,7 @@ func TestReplyTargetCarriesEveryMessageRefField(t *testing.T) { ref := bridge.MessageRef{ RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", AuthorID: "+15551234567", + Outgoing: true, SentAt: time.UnixMilli(1_700_000_000_123), Text: "quoted", HasAttachment: true, diff --git a/internal/bridgeadapters/signal/signal.go b/internal/bridgeadapters/signal/signal.go index d5ce987c..163a99f0 100644 --- a/internal/bridgeadapters/signal/signal.go +++ b/internal/bridgeadapters/signal/signal.go @@ -26,7 +26,7 @@ type poller interface { ObserveIngress(func(account string, line []byte, resolvedSource string, resolvedDestination string)) func() StartPoller(context.Context) (signallive.PollerRun, error) SendTextRequest(string, string, signallive.ReplyTarget) (int64, error) - SendReactionRequest(string, string, string, string, string) error + SendReactionRequest(string, signallive.ReactionTarget, string, string) 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 @@ -163,6 +163,7 @@ func ReplyTarget(ref *bridge.MessageRef) signallive.ReplyTarget { return signallive.ReplyTarget{ RemoteID: ref.RemoteID, AuthorID: ref.AuthorID, + Outgoing: ref.Outgoing, SentAt: ref.SentAt, Text: ref.Text, HasAttachment: ref.HasAttachment, @@ -170,6 +171,19 @@ func ReplyTarget(ref *bridge.MessageRef) signallive.ReplyTarget { } } +// ReactionTarget carries a durable reaction's target to signallive. Signal +// names the target by author and sent timestamp, so the dispatcher's +// description (author, direction, occurred time) travels with the remote ID; +// signallive decides whether those name a message (ReactionTargetArgs). +func ReactionTarget(ref bridge.MessageRef) signallive.ReactionTarget { + return signallive.ReactionTarget{ + RemoteID: ref.RemoteID, + AuthorID: ref.AuthorID, + Outgoing: ref.Outgoing, + SentAt: ref.SentAt, + } +} + // 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( @@ -204,8 +218,7 @@ func (a *Adapter) SendReaction( err := a.poller.SendReactionRequest( req.Conversation.RemoteID, - req.Target.RemoteID, - req.Target.AuthorID, + ReactionTarget(req.Target), req.Emoji, string(req.Action), ) diff --git a/internal/bridgeadapters/signal/signal_test.go b/internal/bridgeadapters/signal/signal_test.go index a1784de6..55ba75a6 100644 --- a/internal/bridgeadapters/signal/signal_test.go +++ b/internal/bridgeadapters/signal/signal_test.go @@ -433,8 +433,9 @@ func TestSendReactionMapsAllActionsAndReturnsEmptyResult(t *testing.T) { AccountID: "signal-primary", Conversation: bridge.ConversationRef{RemoteID: "signal-group:test-group"}, Target: bridge.MessageRef{ - RemoteID: "1700000000123", + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_123), }, Emoji: "👍", Action: action, @@ -447,9 +448,13 @@ func TestSendReactionMapsAllActionsAndReturnsEmptyResult(t *testing.T) { t.Fatalf("SendReaction() result = %+v, want empty ConfirmWithoutResult mapping", result) } request := poller.lastReactionRequest() + wantTarget := signallive.ReactionTarget{ + RemoteID: "f48818f15483f503bb133d92360ca8e2fbd8287e", + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1_700_000_000_123), + } if request.conversationID != "signal-group:test-group" || - request.targetRemoteID != "1700000000123" || - request.targetAuthorID != "+15551234567" || + request.target != wantTarget || request.emoji != "👍" || request.action != string(action) { t.Fatalf("retained SendReactionRequest = %+v, want mapped ReactionRequest", request) @@ -1530,8 +1535,7 @@ type fakeTextRequest struct { type fakeReactionRequest struct { conversationID string - targetRemoteID string - targetAuthorID string + target signallive.ReactionTarget emoji string action string } @@ -1681,13 +1685,14 @@ func (p *fakePoller) lastTextRequest() fakeTextRequest { } func (p *fakePoller) SendReactionRequest( - conversationID, targetRemoteID, targetAuthorID, emoji, action string, + conversationID string, + target signallive.ReactionTarget, + emoji, action string, ) error { p.mu.Lock() p.reactionCalls = append(p.reactionCalls, fakeReactionRequest{ conversationID: conversationID, - targetRemoteID: targetRemoteID, - targetAuthorID: targetAuthorID, + target: target, emoji: emoji, action: action, }) diff --git a/internal/importer/signal_desktop.go b/internal/importer/signal_desktop.go index 4874306d..1d943806 100644 --- a/internal/importer/signal_desktop.go +++ b/internal/importer/signal_desktop.go @@ -24,6 +24,7 @@ import ( "golang.org/x/crypto/pbkdf2" "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/v2keys" ) var ( @@ -377,6 +378,15 @@ func (s *SignalDesktop) openMessageMessage(supportDir string, rawConvo signalDes } messageID, sourceID := signalDesktopMessageIDs(conversationID, senderAddress, timestamp, body, isOutgoing) + if !isOutgoing && row.SentAt <= 0 { + // The Desktop row has no sent time, so timestamp is when it was + // received. Signal names a message by its sender and sent timestamp; + // a hash of the received time is not that name, and a reaction sent to + // it would target nothing. The source ID says so. The message ID stays + // as it was, so a re-import rewrites the row an earlier import stored + // under the unmarked hash instead of adding a second one. + sourceID = v2keys.SignalReceivedSourceID(conversationID, senderAddress, timestamp) + } msg := &db.Message{ MessageID: messageID, ConversationID: conversationID, diff --git a/internal/importer/signal_desktop_reaction_test.go b/internal/importer/signal_desktop_reaction_test.go new file mode 100644 index 00000000..3e1c7dee --- /dev/null +++ b/internal/importer/signal_desktop_reaction_test.go @@ -0,0 +1,266 @@ +package importer + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "slices" + "sync" + "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/messaging" + "github.com/maxghenis/openmessage/internal/migration" + "github.com/maxghenis/openmessage/internal/signallive" + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" +) + +const ( + desktopReactionAccount = "+15550001111" + desktopReactionPeer = "+15551234567" + desktopReactionConversation = "signal:" + desktopReactionPeer + desktopReactionSentAt = int64(1_700_000_000_123) + desktopReactionReceivedAt = int64(1_700_000_000_500) +) + +// stubSignalDesktopExport makes ImportFromDB read exported instead of running +// the Node helper against a Signal Desktop database, and returns the support +// directory to import from. +func stubSignalDesktopExport(t *testing.T, exported signalDesktopExport) string { + t.Helper() + supportDir := filepath.Join(t.TempDir(), "Signal") + if err := os.MkdirAll(filepath.Join(supportDir, "sql"), 0o700); err != nil { + t.Fatalf("mkdir sql dir: %v", err) + } + if err := os.WriteFile(filepath.Join(supportDir, "sql", "db.sqlite"), []byte("stub"), 0o600); err != nil { + t.Fatalf("write stub db: %v", err) + } + raw, err := json.Marshal(exported) + if err != nil { + t.Fatalf("marshal export: %v", err) + } + originalRunHelper := runSignalDesktopHelper + originalNodePath := signalDesktopNodePathFn + originalAddonPath := signalDesktopAddonPathFn + originalSQLKey := signalDesktopSQLKeyFn + originalWriteHelper := writeSignalDesktopHelperFn + t.Cleanup(func() { + runSignalDesktopHelper = originalRunHelper + signalDesktopNodePathFn = originalNodePath + signalDesktopAddonPathFn = originalAddonPath + signalDesktopSQLKeyFn = originalSQLKey + writeSignalDesktopHelperFn = originalWriteHelper + }) + runSignalDesktopHelper = func(context.Context, string, string, ...string) ([]byte, error) { return raw, nil } + signalDesktopNodePathFn = func() (string, error) { return "/usr/bin/node", nil } + signalDesktopAddonPathFn = func() (string, error) { return "/tmp/addon.node", nil } + signalDesktopSQLKeyFn = func(string) (string, error) { return "deadbeef", nil } + writeSignalDesktopHelperFn = func() (string, func(), error) { return "/tmp/helper.cjs", func() {}, nil } + return supportDir +} + +func desktopReactionExport() signalDesktopExport { + return signalDesktopExport{ + Conversations: []signalDesktopConversationRow{{ + ID: "priv-peer", Type: "private", ProfileFullName: "Peer", E164: desktopReactionPeer, ActiveAt: desktopReactionReceivedAt, + }}, + Messages: []signalDesktopMessageRow{ + { + ID: "with-sent-time", ConversationID: "priv-peer", Type: "incoming", Body: "sent time known", + SentAt: desktopReactionSentAt, ReceivedAt: desktopReactionSentAt + 900, Source: desktopReactionPeer, + }, + { + // The fallback row: Signal Desktop recorded no sent time, so + // the importer stores the time it was received. + ID: "no-sent-time", ConversationID: "priv-peer", Type: "incoming", Body: "sent time unknown", + SentAt: 0, ReceivedAt: desktopReactionReceivedAt, Source: desktopReactionPeer, + }, + }, + } +} + +func importDesktopReactionFixture(t *testing.T, store *db.Store) *ImportResult { + t.Helper() + supportDir := stubSignalDesktopExport(t, desktopReactionExport()) + result, err := (&SignalDesktop{ + SupportDir: supportDir, MyName: "Me", MyAddress: desktopReactionAccount, SinceMS: -1, + }).ImportFromDB(store) + if err != nil || len(result.Errors) != 0 { + t.Fatalf("ImportFromDB = %+v, %v", result, err) + } + return result +} + +// reactionRecorder is a scripted Signal account that also accepts reactions, +// recording each request. +type reactionRecorder struct { + *scripted.Adapter + + mu sync.Mutex + requests []bridge.ReactionRequest +} + +func (a *reactionRecorder) SendReaction(_ context.Context, req bridge.ReactionRequest) (bridge.SendResult, error) { + a.mu.Lock() + a.requests = append(a.requests, req) + a.mu.Unlock() + return bridge.SendResult{}, nil +} + +// TestSignalDesktopRowWithoutASentTimeCannotBeAReactionTarget follows two +// incoming Signal Desktop rows from the importer through the migration and the +// v2 reaction dispatcher into the Signal transport's argument builder. A row +// with a sent time is named by it. A row without one is stored under the time +// it was received, which is not the timestamp Signal knows the message by, so +// the transport must refuse it rather than send that time as the target. +func TestSignalDesktopRowWithoutASentTimeCannotBeAReactionTarget(t *testing.T) { + root := t.TempDir() + legacyPath := filepath.Join(root, "messages.db") + legacy, err := db.New(legacyPath) + if err != nil { + t.Fatalf("db.New: %v", err) + } + if result := importDesktopReactionFixture(t, legacy); result.MessagesImported != 2 { + t.Fatalf("ImportFromDB = %+v, want both rows imported", result) + } + + // What the importer wrote. + sentHash := v2keys.SignalIncomingSourceID(desktopReactionConversation, desktopReactionPeer, desktopReactionSentAt) + receivedHash := v2keys.SignalIncomingSourceID(desktopReactionConversation, desktopReactionPeer, desktopReactionReceivedAt) + receivedSourceID := v2keys.SignalReceivedSourceID(desktopReactionConversation, desktopReactionPeer, desktopReactionReceivedAt) + withSent, err := legacy.GetMessageByID("signal:" + sentHash) + if err != nil || withSent == nil || withSent.SourceID != sentHash || withSent.TimestampMS != desktopReactionSentAt { + t.Fatalf("row with a sent time = %+v, %v; want it keyed by its sent timestamp", withSent, err) + } + withoutSent, err := legacy.GetMessageByID("signal:" + receivedHash) + if err != nil || withoutSent == nil || withoutSent.SourceID != receivedSourceID || + withoutSent.TimestampMS != desktopReactionReceivedAt || withoutSent.IsFromMe { + t.Fatalf("row without a sent time = %+v, %v; want source ID %q at the received time", withoutSent, err, receivedSourceID) + } + if err := legacy.Close(); err != nil { + t.Fatalf("close legacy store: %v", err) + } + + staged := filepath.Join(root, "staged.sqlite3") + if _, err := migration.Transform(context.Background(), migration.Options{ + SourcePath: legacyPath, + TempStorePath: staged, + TempBlobPath: filepath.Join(root, "staged-blobs"), + TargetPath: filepath.Join(root, "target"), + TargetStorePath: filepath.Join(root, "target", "store.sqlite3"), + Check: true, + }); err != nil { + t.Fatalf("migration.Transform: %v", err) + } + v2, err := sqlite.Open(staged) + if err != nil { + t.Fatalf("open migrated store: %v", err) + } + t.Cleanup(func() { _ = v2.Close() }) + messages, err := sqlite.NewMessageRepository(v2, time.Now) + if err != nil { + t.Fatalf("NewMessageRepository: %v", err) + } + const accountID = "signal-primary" + conversation, err := v2.GetConversationByRemote(accountID, desktopReactionConversation) + if err != nil { + t.Fatalf("migrated conversation: %v", err) + } + + signal := &reactionRecorder{Adapter: scripted.New(accountID, bridge.PlatformSignal)} + registry := bridge.NewRegistry() + if err := registry.Register(signal); err != nil { + t.Fatalf("Register: %v", err) + } + service, err := messaging.NewMessageService(v2, registry, nil, messaging.SystemClock{}, messaging.CryptoIDSource{}) + if err != nil { + t.Fatalf("NewMessageService: %v", err) + } + + react := func(remoteID string) (bridge.MessageRef, []string, error) { + t.Helper() + target, err := messages.GetMessageByRemote(context.Background(), accountID, conversation.ConversationID, remoteID) + if err != nil { + t.Fatalf("migrated message %q: %v", remoteID, err) + } + if _, err := service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: accountID, ConversationID: conversation.ConversationID, IdempotencyKey: "react-" + remoteID, + }, + TargetMessageID: target.MessageID, + Emoji: "👍", + }); err != nil { + t.Fatalf("SendReaction(%q): %v", remoteID, err) + } + before := len(signal.requests) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(%q) = %d, %v; want 1, nil", remoteID, processed, err) + } + if len(signal.requests) != before+1 { + t.Fatalf("reaction to %q reached the transport %d times, want once", remoteID, len(signal.requests)-before) + } + ref := signal.requests[before].Target + args, err := signallive.ReactionTargetArgs(signaladapter.ReactionTarget(ref), desktopReactionAccount) + return ref, args, err + } + + ref, args, err := react(sentHash) + want := []string{"-a", desktopReactionPeer, "-t", "1700000000123"} + if err != nil || !slices.Equal(args, want) { + t.Fatalf("reaction to the row with a sent time = %q, %v (ref %+v); want %q", args, err, ref, want) + } + + ref, args, err = react(receivedSourceID) + if ref.AuthorID != desktopReactionPeer || ref.Outgoing || !ref.SentAt.Equal(time.UnixMilli(desktopReactionReceivedAt)) { + t.Fatalf("ref for the row without a sent time = %+v, want the peer's incoming message at the received time", ref) + } + if err == nil || err.Error() != "signal reaction target timestamp is unavailable" || args != nil { + t.Fatalf("reaction to the row without a sent time = %q, %v; want it refused, not sent with the received time", args, err) + } +} + +// A row an earlier import stored under the unmarked hash of its received time +// is rewritten in place by the next import, not duplicated, so a legacy store +// stops presenting it as a message keyed by a sent timestamp. +func TestSignalDesktopReimportMarksARowStoredBeforeTheMarker(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() + receivedHash := v2keys.SignalIncomingSourceID(desktopReactionConversation, desktopReactionPeer, desktopReactionReceivedAt) + if err := store.UpsertConversation(&db.Conversation{ + ConversationID: desktopReactionConversation, Name: "Peer", SourcePlatform: "signal", + LastMessageTS: desktopReactionReceivedAt, Participants: "[]", + }); err != nil { + t.Fatalf("UpsertConversation: %v", err) + } + if err := store.UpsertMessage(&db.Message{ + MessageID: "signal:" + receivedHash, ConversationID: desktopReactionConversation, + SenderName: "Peer", SenderNumber: desktopReactionPeer, Body: "sent time unknown", + TimestampMS: desktopReactionReceivedAt, Status: "received", + SourcePlatform: "signal", SourceID: receivedHash, + }); err != nil { + t.Fatalf("seed the row as the earlier importer wrote it: %v", err) + } + + if result := importDesktopReactionFixture(t, store); result.MessagesImported != 1 || result.MessagesDuplicate != 1 { + t.Fatalf("ImportFromDB = %+v, want one new row and one already stored", result) + } + + rows, err := store.GetMessagesByConversation(desktopReactionConversation, 10) + if err != nil || len(rows) != 2 { + t.Fatalf("messages after re-import = %d, %v; want the two Desktop rows and no duplicate", len(rows), err) + } + row, err := store.GetMessageByID("signal:" + receivedHash) + want := v2keys.SignalReceivedSourceID(desktopReactionConversation, desktopReactionPeer, desktopReactionReceivedAt) + if err != nil || row == nil || row.SourceID != want { + t.Fatalf("re-imported row = %+v, %v; want source ID %q", row, err, want) + } +} diff --git a/internal/messaging/dispatch.go b/internal/messaging/dispatch.go index 2adca45c..85606977 100644 --- a/internal/messaging/dispatch.go +++ b/internal/messaging/dispatch.go @@ -744,6 +744,7 @@ func (s *MessageService) targetRefForLease( return bridge.MessageRef{ RemoteID: target.RemoteMessageID, AuthorID: authorID, + Outgoing: target.Direction == sqlite.MessageDirectionOutgoing, SentAt: time.UnixMilli(target.OccurredAtMS), }, "", nil } diff --git a/internal/messaging/reaction_ref_test.go b/internal/messaging/reaction_ref_test.go new file mode 100644 index 00000000..5397a44d --- /dev/null +++ b/internal/messaging/reaction_ref_test.go @@ -0,0 +1,357 @@ +package messaging + +import ( + "bytes" + "context" + "errors" + "fmt" + "math/rand" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +// notDispatchedReaction is what a transport answers when it cannot name the +// target yet (Signal: no sent timestamp): nothing was sent, retry later. +func notDispatchedReaction() sendStep { + return sendStep{err: bridge.OpError{ + Class: bridge.FailureTransient, Operation: reactionOperation, + Dispatch: bridge.DispatchNotCalled, Cause: errors.New("scripted: target has no transport timestamp"), + }} +} + +// A transport that needs to know whose message a reaction targets cannot read +// that off AuthorID: it is empty for this account's sends and for an incoming +// message stored with no sender alike. +func TestReactionTargetRefTellsAnOwnMessageFromASenderlessIncomingOne(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + senderless := mustProjectDispatchMessage(t, store, clock, sqlite.Message{ + MessageID: "message-senderless-incoming", + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: "1700000000123", + Direction: sqlite.MessageDirectionIncoming, + Body: "group message stored with no source", + State: sqlite.MessageStateActive, + OccurredAtMS: 1_700_000_000_123, + }) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "1700000000555"}}}} + reactions := &scriptedReactionSender{steps: []sendStep{{}, {}}} + registry := newScriptedRegistry("reaction-direction", text) + registry.setReactionSender(reactions) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + own := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reaction-direction-own"), + Body: "own send", + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(own send) = %d, %v; want 1, nil", processed, err) + } + + for index, targetID := range []string{senderless.MessageID, own.LocalMessageID} { + mustSendDispatchReaction(t, service, SendReactionCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("reaction-direction-%d", index)), + TargetMessageID: targetID, + Emoji: "👍", + }) + clock.Advance(time.Millisecond) + } + if processed, err := service.DispatchDue(context.Background(), 2); err != nil || processed != 2 { + t.Fatalf("DispatchDue(reactions) = %d, %v; want 2, nil", processed, err) + } + requests := reactions.snapshotRequests() + if len(requests) != 2 { + t.Fatalf("reaction requests = %+v, want two", requests) + } + wantIncoming := bridge.MessageRef{RemoteID: "1700000000123", SentAt: time.UnixMilli(1_700_000_000_123)} + wantOwn := bridge.MessageRef{ + RemoteID: "1700000000555", Outgoing: true, SentAt: time.UnixMilli(messagingTestTime.UnixMilli()), + } + if !targetRefEqual(requests[0].Target, wantIncoming) { + t.Fatalf("senderless incoming target = %+v, want %+v", requests[0].Target, wantIncoming) + } + if !targetRefEqual(requests[1].Target, wantOwn) { + t.Fatalf("own target = %+v, want %+v", requests[1].Target, wantOwn) + } +} + +// A reaction to this account's own send that the transport has not accepted +// carries the send's request ID. Once the send is confirmed, a retry carries +// the transport's ID for it. +func TestReactionToOwnPendingSendFollowsItsConfirmation(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{steps: []sendStep{{result: bridge.SendResult{RemoteMessageID: "1700000000555"}}}} + reactions := &scriptedReactionSender{steps: []sendStep{notDispatchedReaction(), {}}} + registry := newScriptedRegistry("reaction-own-pending", text) + registry.setReactionSender(reactions) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + // The reaction is due now; its target is scheduled a little later, so the + // first reaction attempt finds a message the transport has not accepted. + ownCommand := testCommonCommand("reaction-own-pending-send") + ownCommand.NotBefore = clock.Now().Add(2 * time.Second) + own := mustSendText(t, service, SendTextCommand{CommonCommand: ownCommand, Body: "scheduled"}) + ownRow := mustOutboxItem(t, service, own.OutboxID) + reaction := mustSendDispatchReaction(t, service, SendReactionCommand{ + CommonCommand: testCommonCommand("reaction-own-pending"), + TargetMessageID: own.LocalMessageID, + Emoji: "👍", + }) + + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reaction before its target) = %d, %v; want 1, nil", processed, err) + } + submittedAt := time.UnixMilli(messagingTestTime.UnixMilli()) + requests := reactions.snapshotRequests() + wantPending := bridge.MessageRef{RemoteID: ownRow.TransportRequestID, Outgoing: true, SentAt: submittedAt} + if len(requests) != 1 || !targetRefEqual(requests[0].Target, wantPending) { + t.Fatalf("reaction target while the send was pending = %+v, want %+v", requests, wantPending) + } + if row := mustOutboxItem(t, service, reaction.OutboxID); row.State != sqlite.OutboxNotDispatched { + t.Fatalf("reaction row = %+v, want not_dispatched awaiting its retry", row) + } + + clock.Advance(3 * time.Second) + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(own send) = %d, %v; want 1, nil", processed, err) + } + clock.Advance(defaultRetryDelay) + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reaction retry) = %d, %v; want 1, nil", processed, err) + } + requests = reactions.snapshotRequests() + wantConfirmed := bridge.MessageRef{RemoteID: "1700000000555", Outgoing: true, SentAt: submittedAt} + if len(requests) != 2 || !targetRefEqual(requests[1].Target, wantConfirmed) { + t.Fatalf("reaction target after the send confirmed = %+v, want %+v", requests, wantConfirmed) + } + if delivery := mustDelivery(t, service, reaction.OutboxID); delivery.State != OutboxConfirmed { + t.Fatalf("reaction delivery = %+v, want confirmed", delivery) + } +} + +func TestDescribedReplyRefStatesWhoSentTheQuotedMessage(t *testing.T) { + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + incoming := projectReplyTarget(t, store, clock, sqlite.Message{ + MessageID: "message-quoted-senderless", + ConversationID: "conversation-1", + AccountID: "account-1", + RemoteMessageID: "4f2b7c9d0e1a3b5c7d9e1f2a4b6c8d0e2f4a6b8c", + Direction: sqlite.MessageDirectionIncoming, + Body: "incoming", + State: sqlite.MessageStateActive, + OccurredAtMS: 1_700_000_000_123, + }) + text := &scriptedTextSender{steps: []sendStep{ + {result: bridge.SendResult{RemoteMessageID: "1700000000555"}}, + {result: bridge.SendResult{RemoteMessageID: "1700000000556"}}, + {result: bridge.SendResult{RemoteMessageID: "1700000000557"}}, + }} + registry := newScriptedRegistry("reply-direction", text) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + own := mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand("reply-direction-own"), + Body: "own send", + }) + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(own send) = %d, %v; want 1, nil", processed, err) + } + for index, quotedID := range []string{incoming.MessageID, own.LocalMessageID} { + clock.Advance(time.Millisecond) + mustSendText(t, service, SendTextCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("reply-direction-%d", index)), + Body: "reply", + ReplyToMessageID: quotedID, + }) + } + if processed, err := service.DispatchDue(context.Background(), 2); err != nil || processed != 2 { + t.Fatalf("DispatchDue(replies) = %d, %v; want 2, nil", processed, err) + } + requests := text.snapshotRequests() + if len(requests) != 3 || requests[1].ReplyTo == nil || requests[2].ReplyTo == nil { + t.Fatalf("text requests = %+v, want the own send and two replies", requests) + } + if requests[1].ReplyTo.Outgoing || requests[1].ReplyTo.AuthorID != "" { + t.Fatalf("reply ref to a senderless incoming message = %+v, want no author and not outgoing", requests[1].ReplyTo) + } + if !requests[2].ReplyTo.Outgoing || requests[2].ReplyTo.AuthorID != "" { + t.Fatalf("reply ref to an own send = %+v, want outgoing", requests[2].ReplyTo) + } +} + +// TestReactionTargetRefInvariants checks, over seeded random conversations, +// the target ref of every reaction the dispatcher maps: +// +// - The ref names the target message's current remote ID: its outbox +// request ID until the transport confirms the send, then the transport's. +// - Outgoing is set exactly for a message this account sent, whether or not +// the stored message names a sender. +// - AuthorID is the sender's canonical value, "" when the message names +// none, and SentAt is the occurred time. +// - The reply-only fields (Text and the attachment fields) stay empty. +// - The mapping is deterministic: a retried attempt carries the same ref. +func TestReactionTargetRefInvariants(t *testing.T) { + for seed := int64(1); seed <= 40; seed++ { + seed := seed + t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { + checkReactionTargetRefInvariants(t, seed) + }) + } +} + +type reactionRefCase struct { + name string + targetID string // local message ID the reaction targets + want bridge.MessageRef +} + +func checkReactionTargetRefInvariants(t *testing.T, seed int64) { + random := rand.New(rand.NewSource(seed)) + clock := newManualClock(messagingTestTime) + store := openMessagingTestStore(t, clock.Now()) + text := &scriptedTextSender{} + media := &scriptedMediaSender{} + reactions := &scriptedReactionSender{} + registry := newScriptedRegistry(bridge.Platform(fmt.Sprintf("reaction-invariants-%d", seed)), text) + registry.setMediaSender(media) + registry.setReactionSender(reactions) + registry.setAvailable(true) + service := newMessagingTestService(t, store, registry, clock) + + var cases []reactionRefCase + 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", + Direction: sqlite.MessageDirectionIncoming, + Body: "incoming body is not copied", + State: sqlite.MessageStateActive, + OccurredAtMS: 1_700_000_000_000 + random.Int63n(1_000_000_000), + } + // Incoming IDs are SHA-1s from the live decoder and can be decimal + // for a migrated row. + if random.Intn(3) == 0 { + message.RemoteMessageID = fmt.Sprintf("%d", 1_600_000_000_000+int64(index)) + } else { + message.RemoteMessageID = fmt.Sprintf("%040x", random.Uint64()) + } + author := "" + if random.Intn(3) != 0 { + identity := random.Intn(identityCount) + identityID := fmt.Sprintf("identity-%d", identity) + message.SenderIdentityID = &identityID + author = fmt.Sprintf("+1555000%04d", identity) + } + var attachmentMIMEs []string + if random.Intn(2) == 0 { + attachmentMIMEs = []string{"image/jpeg"} + } + projectReplyTarget(t, store, clock, message, attachmentMIMEs...) + cases = append(cases, reactionRefCase{ + name: message.MessageID, targetID: message.MessageID, + want: bridge.MessageRef{ + RemoteID: message.RemoteMessageID, AuthorID: author, SentAt: time.UnixMilli(message.OccurredAtMS), + }, + }) + } + + outgoingCount := 1 + random.Intn(5) + for index := 0; index < outgoingCount; index++ { + occurredMS := clock.Now().UnixMilli() + // 0: confirmed by the transport, 1: canceled before sending, + // 2: scheduled far in the future, so still unsent. + fate := random.Intn(3) + command := testCommonCommand(fmt.Sprintf("outgoing-%d", index)) + if fate == 2 { + command.NotBefore = clock.Now().Add(1000 * time.Hour) + } + isMedia := random.Intn(3) == 0 + var submission Submission + if isMedia { + submission = mustSendMedia(t, service, SendMediaCommand{ + CommonCommand: command, + Content: bytes.NewReader([]byte(fmt.Sprintf("bytes-%d", index))), + Filename: "attachment.bin", + MIME: "image/png", + }) + } else { + submission = mustSendText(t, service, SendTextCommand{CommonCommand: command, Body: "outgoing text"}) + } + row := mustOutboxItem(t, service, submission.OutboxID) + wantRemote := row.TransportRequestID + switch fate { + case 0: + wantRemote = fmt.Sprintf("%d", 1_700_000_000_000+int64(index)) + step := sendStep{result: bridge.SendResult{RemoteMessageID: wantRemote}} + if isMedia { + media.mu.Lock() + media.steps = append(media.steps, step) + media.mu.Unlock() + } else { + text.mu.Lock() + text.steps = append(text.steps, step) + 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) + } + case 1: + if _, err := service.Cancel(context.Background(), submission.OutboxID); err != nil { + t.Fatalf("seed %d: Cancel(outgoing %d): %v", seed, index, err) + } + } + cases = append(cases, reactionRefCase{ + name: submission.LocalMessageID, targetID: submission.LocalMessageID, + want: bridge.MessageRef{RemoteID: wantRemote, Outgoing: true, SentAt: time.UnixMilli(occurredMS)}, + }) + clock.Advance(time.Second) + } + + random.Shuffle(len(cases), func(i, j int) { cases[i], cases[j] = cases[j], cases[i] }) + for index, tc := range cases { + // The first attempt fails before dispatch, so the target is mapped twice. + reactions.mu.Lock() + reactions.steps = append(reactions.steps, notDispatchedReaction(), sendStep{}) + before := len(reactions.requests) + reactions.mu.Unlock() + mustSendDispatchReaction(t, service, SendReactionCommand{ + CommonCommand: testCommonCommand(fmt.Sprintf("reaction-%d", index)), + TargetMessageID: tc.targetID, + Emoji: "👍", + }) + 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, tc.name, attempt, processed, err) + } + clock.Advance(defaultRetryDelay + time.Second) + } + requests := reactions.snapshotRequests()[before:] + if len(requests) != 2 { + t.Fatalf("seed %d case %s: reaction attempts = %d, want 2", seed, tc.name, len(requests)) + } + if !targetRefEqual(requests[0].Target, requests[1].Target) { + t.Fatalf("seed %d case %s: retried target %+v != first %+v", seed, tc.name, requests[1].Target, requests[0].Target) + } + if got := requests[1].Target; !targetRefEqual(got, tc.want) { + t.Fatalf("seed %d case %s: target = %+v, want %+v", seed, tc.name, got, tc.want) + } + } +} + +// targetRefEqual compares every field of two refs, the direction included. +func targetRefEqual(got, want bridge.MessageRef) bool { + return got.Outgoing == want.Outgoing && messageRefEqual(&got, &want) +} diff --git a/internal/messaging/reply_ref.go b/internal/messaging/reply_ref.go index c8dd3dd8..ecbeed6e 100644 --- a/internal/messaging/reply_ref.go +++ b/internal/messaging/reply_ref.go @@ -62,6 +62,7 @@ func (s *MessageService) replyRefForLease( return &bridge.MessageRef{ RemoteID: target.RemoteMessageID, AuthorID: authorID, + Outgoing: target.Direction == sqlite.MessageDirectionOutgoing, SentAt: time.UnixMilli(target.OccurredAtMS), Text: target.Body, HasAttachment: hasAttachment, diff --git a/internal/signallive/client.go b/internal/signallive/client.go index e6ceedf6..f46352f4 100644 --- a/internal/signallive/client.go +++ b/internal/signallive/client.go @@ -1390,14 +1390,17 @@ func signalSendAllRecipientsFailed(err error) bool { return errors.As(err, &resultErr) && resultErr.allRecipientsFailed } -// SendReactionRequest sends a reaction using only the durable request's remote -// references. Unlike the legacy SendReaction path, it neither reads nor updates -// locally stored message state. +// SendReactionRequest sends a reaction using only the durable request's +// description of its target. Unlike the legacy SendReaction path, it neither +// reads nor updates locally stored message state. The target's author and +// sent timestamp come from ReactionTargetArgs; a target it cannot name fails +// before signal-cli runs. func (b *Bridge) SendReactionRequest( - conversationID, targetRemoteID, targetAuthorID, emoji, action string, + conversationID string, + target ReactionTarget, + emoji, action string, ) error { - targetRemoteID = strings.TrimSpace(targetRemoteID) - if targetRemoteID == "" { + if strings.TrimSpace(target.RemoteID) == "" { return errors.New("signal target message is required") } emoji = strings.TrimSpace(emoji) @@ -1417,19 +1420,16 @@ func (b *Bridge) SendReactionRequest( if err != nil { return err } - targetAuthor := strings.TrimSpace(targetAuthorID) - if targetAuthor == "" { - // An empty author identifies a message sent by this account. - targetAuthor = account + targetArgs, err := ReactionTargetArgs(target, account) + if err != nil { + return err } - args := []string{ + args := append([]string{ "-a", account, "sendReaction", "-e", emoji, - "-a", targetAuthor, - "-t", targetRemoteID, - } + }, targetArgs...) if action == "remove" { args = append(args, "-r") } diff --git a/internal/signallive/client_test.go b/internal/signallive/client_test.go index 6088a0ad..13b8cf21 100644 --- a/internal/signallive/client_test.go +++ b/internal/signallive/client_test.go @@ -19,6 +19,7 @@ import ( "github.com/maxghenis/openmessage/internal/db" "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" ) func TestQRCodeRendersDataURL(t *testing.T) { @@ -2248,10 +2249,15 @@ func TestBridgeSendReactionRequestMapsActionsWithoutStoredState(t *testing.T) { return []byte("ok"), nil } + // An incoming target: its v2 remote ID is a SHA-1, so -t is the + // described sent time. err := bridge.SendReactionRequest( "signal:+15551234567", - "1700000000123", - "+15551234567", + ReactionTarget{ + RemoteID: v2keys.SignalIncomingSourceID("signal:+15551234567", "+15551234567", 1700000000123), + AuthorID: "+15551234567", + SentAt: time.UnixMilli(1700000000123), + }, "😂", tc.action, ) @@ -2295,10 +2301,11 @@ func TestBridgeSendReactionRequestMapsEmptyAuthorToSelfForGroup(t *testing.T) { return []byte("ok"), nil } + // This account's confirmed send: its remote ID is the timestamp signal-cli + // reported, while its occurred time is the earlier submit time. if err := bridge.SendReactionRequest( "signal-group:test-group", - "1700000000123", - "", + ReactionTarget{RemoteID: "1700000000123", Outgoing: true, SentAt: time.UnixMilli(1700000000000)}, "👍", "switch", ); err != nil { @@ -2333,7 +2340,8 @@ func TestBridgeSendReactionRequestSeparatesPreCallAndCommandFailures(t *testing. return nil, errors.New("exit status 1") } - err := bridge.SendReactionRequest("invalid-conversation", "1700000000123", "", "👍", "add") + ownSend := ReactionTarget{RemoteID: "1700000000123", Outgoing: true} + err := bridge.SendReactionRequest("invalid-conversation", ownSend, "👍", "add") if err == nil || IsCommandError(err) || IsSendNotDispatchedError(err) { t.Fatalf("pre-call error = %v (%T), want unmarked validation error", err, err) } @@ -2341,7 +2349,7 @@ func TestBridgeSendReactionRequestSeparatesPreCallAndCommandFailures(t *testing. t.Fatalf("signal-cli calls after pre-call failure = %d, want 0", calls) } - err = bridge.SendReactionRequest("signal:+15551234567", "1700000000123", "", "👍", "add") + err = bridge.SendReactionRequest("signal:+15551234567", ownSend, "👍", "add") if err == nil || !IsCommandError(err) || IsSendNotDispatchedError(err) { t.Fatalf("send-boundary error = %v (%T), want uncertain CommandError", err, err) } diff --git a/internal/signallive/export_test.go b/internal/signallive/export_test.go index c80c3b87..04523f99 100644 --- a/internal/signallive/export_test.go +++ b/internal/signallive/export_test.go @@ -74,6 +74,43 @@ func LegacyQuoteArgsForTest( return bridge.signalQuoteArgs(replyToID, account) } +// NewConnectedBridgeForTest is a Bridge that reports account as paired and +// connected, so the Signal adapter dispatches to it. Every signal-cli call goes +// through runSignalCLI (see SetRunSignalCLIForTest); store is the legacy store +// the retained legacy paths read. +func NewConnectedBridgeForTest( + store *db.Store, + configDir string, + account string, + contactByACI map[string]string, +) *Bridge { + if contactByACI == nil { + contactByACI = map[string]string{} + } + return &Bridge{ + store: store, + logger: zerolog.Nop(), + configDir: configDir, + account: account, + connected: true, + contactByACI: contactByACI, + } +} + +// LegacySendReactionForTest runs the legacy reaction path (SendReaction, which +// looks its target up in the legacy store by message ID) from a connected +// bridge over store, with whatever runSignalCLI is installed. +func LegacySendReactionForTest( + store *db.Store, + configDir string, + account string, + contactByACI map[string]string, + conversationID, messageID, emoji, action string, +) error { + return NewConnectedBridgeForTest(store, configDir, account, contactByACI). + SendReaction(conversationID, messageID, emoji, action) +} + // ReplyQuoteArgsForTest is the quote SendTextRequest and SendMediaRequest // build for reply. func ReplyQuoteArgsForTest( diff --git a/internal/signallive/reaction_target.go b/internal/signallive/reaction_target.go new file mode 100644 index 00000000..5ad6c2f5 --- /dev/null +++ b/internal/signallive/reaction_target.go @@ -0,0 +1,70 @@ +package signallive + +import ( + "errors" + "strconv" + "strings" + "time" + + "github.com/maxghenis/openmessage/internal/v2keys" +) + +// ReactionTarget is the message a durable Signal reaction targets, as the v2 +// dispatcher describes it (bridge.MessageRef). RemoteID is its v2 remote ID, +// AuthorID its sender's canonical identity (empty when the store names none), +// Outgoing whether this account sent it, and SentAt its stored occurred time. +type ReactionTarget struct { + RemoteID string + AuthorID string + Outgoing bool + SentAt time.Time +} + +// ReactionTargetArgs builds signal-cli sendReaction's target arguments, +// "-a -t ", for a reaction sent from account. +// +// Signal names the message a reaction targets by its author and sent +// timestamp, never by an ID OpenMessage stores. A reaction that names a +// message nobody has is still a well-formed command, so this fails, before +// signal-cli runs, unless the store vouches for both: +// +// - An incoming message is named by its stored sender and its occurred +// time, and only when its remote ID has the form a Signal receiver gives a +// message it keyed by sender and sent timestamp (a SHA-1, +// v2keys.SignalIncomingSourceID): the v2 decoder, the legacy receiver, and +// a Signal Desktop import of a row with a sent time all store that +// timestamp as the occurred time. The ID itself is not sent. Any other ID +// is refused: a Desktop row with no sent time is stored under its received +// time and marked (v2keys.SignalReceivedSourceID), and the legacy-primary +// mirror keeps a legacy ID. A message with no stored sender has no author +// to name. +// - A message this account sent is named by this account and by its remote +// ID, and only when that ID is a decimal timestamp. That is how a send +// Signal accepted is stored when its timestamp was kept: an outbox +// confirmation, a sync message from the phone, the legacy SendMedia and +// the legacy-primary projector all key it by that timestamp. Its occurred +// time is not used, because for every other ID it is not known to be Signal's: +// an outbox send still on its request ID carries its submit time, a +// migrated scheduled send its creation time, and a migrated "local:" row +// may carry the wall clock the legacy SendText read after signal-cli +// returned. +// +// Quotes (QuoteArgs) are more lenient about a sent message's time: a quote +// carries the quoted text with it. +func ReactionTargetArgs(target ReactionTarget, account string) ([]string, error) { + author := normalizeSignalAddress(target.AuthorID) + var timestampMS int64 + if target.Outgoing { + author = normalizeSignalAddress(account) + timestampMS, _ = signalTimestampID(target.RemoteID) + } else if v2keys.IsSignalIncomingSourceID(strings.TrimSpace(target.RemoteID)) && !target.SentAt.IsZero() { + timestampMS = target.SentAt.UnixMilli() + } + if author == "" { + return nil, errors.New("signal reaction target author is unavailable") + } + if timestampMS <= 0 { + return nil, errors.New("signal reaction target timestamp is unavailable") + } + return []string{"-a", author, "-t", strconv.FormatInt(timestampMS, 10)}, nil +} diff --git a/internal/signallive/reaction_target_external_test.go b/internal/signallive/reaction_target_external_test.go new file mode 100644 index 00000000..5c4643d3 --- /dev/null +++ b/internal/signallive/reaction_target_external_test.go @@ -0,0 +1,629 @@ +package signallive_test + +import ( + "cmp" + "context" + "fmt" + "math/rand" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/maxghenis/openmessage/internal/bridge" + 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" +) + +// signalCLIRecorder stands in for signal-cli: it records every sendReaction +// argv, answers a text send with the next scripted timestamp, and answers +// contact refreshes with no contacts. +type signalCLIRecorder struct { + mu sync.Mutex + reactions [][]string + sendTimestamps []int64 +} + +func installSignalCLIRecorder(t *testing.T) *signalCLIRecorder { + t.Helper() + recorder := &signalCLIRecorder{} + restore := signallive.SetRunSignalCLIForTest(func(_ context.Context, _ string, args ...string) ([]byte, error) { + recorder.mu.Lock() + defer recorder.mu.Unlock() + switch { + case slices.Contains(args, "sendReaction"): + recorder.reactions = append(recorder.reactions, slices.Clone(args)) + return []byte("ok"), nil + case slices.Contains(args, "send") && len(recorder.sendTimestamps) > 0: + timestamp := recorder.sendTimestamps[0] + recorder.sendTimestamps = recorder.sendTimestamps[1:] + return []byte(fmt.Sprintf( + `{"timestamp":%d,"results":[{"recipientAddress":{"number":"+15551234567"},"type":"SUCCESS"}]}`, + timestamp, + )), nil + case slices.Contains(args, "listContacts"): + return []byte("[]"), nil + } + return nil, fmt.Errorf("unexpected signal-cli call %q", args) + }) + t.Cleanup(restore) + return recorder +} + +func (r *signalCLIRecorder) reactionCalls() [][]string { + r.mu.Lock() + defer r.mu.Unlock() + return slices.Clone(r.reactions) +} + +func (r *signalCLIRecorder) scriptSend(timestamp int64) { + r.mu.Lock() + r.sendTimestamps = append(r.sendTimestamps, timestamp) + r.mu.Unlock() +} + +// newSignalReactionService wires a messaging service to the real Signal +// adapter over a connected signallive bridge whose legacy store is legacy. +func newSignalReactionService( + t *testing.T, + v2Store *sqlite.Store, + legacy *db.Store, + clock messaging.Clock, +) *messaging.MessageService { + t.Helper() + registry := bridge.NewRegistry() + live := signallive.NewConnectedBridgeForTest(legacy, t.TempDir(), quoteDiffAccount, nil) + if err := registry.Register(signaladapter.New(quoteDiffAccountID, live)); err != nil { + t.Fatalf("Register(Signal adapter): %v", err) + } + service, err := messaging.NewMessageService(v2Store, registry, nil, clock, messaging.CryptoIDSource{}) + if err != nil { + t.Fatalf("NewMessageService(): %v", err) + } + return service +} + +// ingestSignalLine stores one signal-cli receive line through the durable +// tee, v2 decoder and ingest worker, as a v2-primary daemon does. +func ingestSignalLine( + t *testing.T, + sink *ingest.Sink, + worker *ingest.Worker, + messages *sqlite.MessageRepository, + line []byte, + resolvedSource string, + receivedAt time.Time, +) { + t.Helper() + record, ephemeral, err := ingest.BuildSignalIngress( + quoteDiffAccountID, 1, quoteDiffAccount, line, resolvedSource, "", receivedAt, + ) + if err != nil || record == nil || ephemeral != nil { + t.Fatalf("BuildSignalIngress = (%v, %v, %v)", record, ephemeral, err) + } + if err := sink.AppendIngress(context.Background(), *record); err != nil { + t.Fatalf("AppendIngress: %v", err) + } + drainQuoteDiffWorker(t, worker, messages) +} + +// TestV2ReactionToIncomingSignalMessageTargetsItsSentTimestamp dispatches a +// reaction to a live-ingested incoming Signal message through the v2 outbox, +// the real Signal adapter and signallive, and checks the signal-cli argv. +// Signal names the reacted-to message by author and sent timestamp; the v2 +// remote ID of an incoming message is a SHA-1, so it must not reach -t. +func TestV2ReactionToIncomingSignalMessageTargetsItsSentTimestamp(t *testing.T) { + recorder := installSignalCLIRecorder(t) + dir := t.TempDir() + emptyLegacy := openQuoteDiffLegacy(t, filepath.Join(dir, "legacy.sqlite3")) + v2Store, messages, sink, worker := openQuoteDiffV2(t, filepath.Join(dir, "v2.sqlite3")) + + const sender = "+15551234567" + const conversation = "signal:" + sender + sentAt := int64(1_700_000_000_123) + // The envelope timestamp differs from the data message's sent timestamp. + line := []byte(fmt.Sprintf( + `{"account":%q,"envelope":{"source":%q,"sourceNumber":%q,"sourceName":"Taylor","timestamp":%d,"dataMessage":{"timestamp":%d,"message":"react to me"}}}`, + quoteDiffAccount, sender, sender, sentAt+7, sentAt, + )) + ingestSignalLine(t, sink, worker, messages, line, sender, time.UnixMilli(sentAt+7)) + + v2Conversation, err := v2Store.GetConversationByRemote(quoteDiffAccountID, conversation) + if err != nil { + t.Fatalf("GetConversationByRemote(%q): %v", conversation, err) + } + rows, err := messages.ListMessagesByConversation(context.Background(), v2Conversation.ConversationID, 0, "", 10) + if err != nil || len(rows) != 1 { + t.Fatalf("v2 messages = %+v, %v; want the one ingested message", rows, err) + } + target := rows[0] + if want := v2keys.SignalIncomingSourceID(conversation, sender, sentAt); target.RemoteMessageID != want { + t.Fatalf("incoming remote ID = %q, want the SHA-1 %q", target.RemoteMessageID, want) + } + + service := newSignalReactionService(t, v2Store, emptyLegacy, messaging.SystemClock{}) + submission, err := service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: quoteDiffAccountID, + ConversationID: v2Conversation.ConversationID, + IdempotencyKey: "react-to-incoming", + }, + TargetMessageID: target.MessageID, + Emoji: "👍", + Action: bridge.ReactionAdd, + }) + if err != nil { + t.Fatalf("SendReaction(): %v", err) + } + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue() = %d, %v; want 1, nil", processed, err) + } + + calls := recorder.reactionCalls() + want := []string{ + "-a", quoteDiffAccount, + "sendReaction", + "-e", "👍", + "-a", sender, + "-t", "1700000000123", + sender, + } + if len(calls) != 1 || !slices.Equal(calls[0], want) { + t.Fatalf("signal-cli sendReaction calls = %q, want one %q", calls, want) + } + delivery, err := service.Get(context.Background(), submission.OutboxID) + if err != nil || delivery.State != messaging.OutboxConfirmed { + t.Fatalf("reaction delivery = %+v, %v; want confirmed", delivery, err) + } +} + +// TestV2ReactionMatchesLegacyReaction is the differential test for Signal +// reactions on a v2-primary daemon. Each seed feeds one random corpus of +// signal-cli receive lines (the quote differential's eight shapes) through +// both retained paths: the legacy receive handler and the durable tee, v2 +// decoder and ingest worker. For every message both stores then hold it sends +// the same reaction twice and compares the signal-cli argv: +// +// - the legacy path: SendReaction, which looks the target up in the legacy +// store by message ID and sends its stored sender and timestamp; +// - the new path: a reaction submitted to the messaging service with the v2 +// message's ID and dispatched through the real Signal adapter to a bridge +// whose legacy store is empty. +// +// The argv must be identical: every shape here is one both paths can name (a +// decoded incoming message with a sender, or a sync message carrying Signal's +// timestamp). Before the fix the new path sent the v2 remote ID as -t, which +// for every incoming shape is a SHA-1. +func TestV2ReactionMatchesLegacyReaction(t *testing.T) { + recorder := installSignalCLIRecorder(t) + 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 += checkReactionDifferential(t, seed, covered, recorder) + }) + } + for kind, count := range covered { + if count == 0 { + t.Errorf("corpus kind %d was never generated", kind) + } + } + t.Logf("compared %d reactions; per-kind coverage %v", compared, covered) +} + +type reactionDiffKey struct { + conversation string + timestamp int64 + fromMe bool +} + +func checkReactionDifferential(t *testing.T, seed int64, covered []int, recorder *signalCLIRecorder) 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) + + legacyRows := map[reactionDiffKey]*db.Message{} + v2Rows := map[reactionDiffKey]sqlite.Message{} + conversations, err := legacy.ListConversations(1000) + if err != nil { + t.Fatalf("legacy ListConversations: %v", err) + } + 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[reactionDiffKey{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[reactionDiffKey{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)) + } + keys := make([]reactionDiffKey, 0, len(legacyRows)) + for key := range legacyRows { + keys = append(keys, key) + } + slices.SortFunc(keys, func(a, b reactionDiffKey) int { + return cmp.Or( + strings.Compare(a.conversation, b.conversation), + cmp.Compare(a.timestamp, b.timestamp), + ) + }) + + service := newSignalReactionService(t, v2Store, emptyLegacy, messaging.SystemClock{}) + emojis := []string{"👍", "❤️", "😂"} + actions := []bridge.ReactionAction{bridge.ReactionAdd, bridge.ReactionRemove, bridge.ReactionSwitch} + for index, key := range keys { + legacyRow := legacyRows[key] + v2Row, ok := v2Rows[key] + if !ok { + t.Fatalf("v2 store lacks the message legacy holds as %+v", legacyRow) + } + emoji := emojis[random.Intn(len(emojis))] + action := actions[random.Intn(len(actions))] + + before := len(recorder.reactionCalls()) + if err := signallive.LegacySendReactionForTest( + legacy, t.TempDir(), quoteDiffAccount, contacts, + legacyRow.ConversationID, legacyRow.MessageID, emoji, string(action), + ); err != nil { + t.Fatalf("legacy reaction to %+v: %v", legacyRow, err) + } + submission, err := service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: quoteDiffAccountID, + ConversationID: v2Row.ConversationID, + IdempotencyKey: fmt.Sprintf("reaction-diff-%d-%d", seed, index), + }, + TargetMessageID: v2Row.MessageID, + Emoji: emoji, + Action: action, + }) + if err != nil { + t.Fatalf("SendReaction(%q): %v", v2Row.MessageID, err) + } + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reaction to %q) = %d, %v", v2Row.MessageID, processed, err) + } + calls := recorder.reactionCalls()[before:] + if len(calls) != 2 { + t.Fatalf("seed %d: reaction to %+v ran %d sendReaction calls, want legacy then v2: %q", seed, v2Row, len(calls), calls) + } + if !slices.Equal(calls[1], calls[0]) { + t.Fatalf("seed %d: v2 reaction to %+v ran %q; legacy row %+v runs %q", seed, v2Row, calls[1], legacyRow, calls[0]) + } + if delivery, err := service.Get(context.Background(), submission.OutboxID); err != nil || + delivery.State != messaging.OutboxConfirmed { + t.Fatalf("seed %d: v2 reaction delivery = %+v, %v; want confirmed", seed, delivery, err) + } + } + return len(keys) +} + +// TestV2ReactionToOwnPendingSendWaitsForItsTransportTimestamp reacts to this +// account's own send while the send is still scheduled, so the message has no +// Signal timestamp yet: its remote ID is its outbox request ID and its +// occurred time the submit time. The reaction must not reach signal-cli with +// either; it is not dispatched and retries. Once the send is confirmed, its +// remote ID is the timestamp signal-cli reported, and the retry targets it. +func TestV2ReactionToOwnPendingSendWaitsForItsTransportTimestamp(t *testing.T) { + recorder := installSignalCLIRecorder(t) + dir := t.TempDir() + emptyLegacy := openQuoteDiffLegacy(t, filepath.Join(dir, "legacy.sqlite3")) + v2Store, messages, sink, worker := openQuoteDiffV2(t, filepath.Join(dir, "v2.sqlite3")) + + const peer = "+15551234567" + start := time.UnixMilli(1_760_000_000_000) + // One incoming line creates the conversation. + line := []byte(fmt.Sprintf( + `{"account":%q,"envelope":{"source":%q,"sourceNumber":%q,"timestamp":%d,"dataMessage":{"timestamp":%d,"message":"hello"}}}`, + quoteDiffAccount, peer, peer, start.UnixMilli()-60_000, start.UnixMilli()-60_000, + )) + ingestSignalLine(t, sink, worker, messages, line, peer, start.Add(-time.Minute)) + v2Conversation, err := v2Store.GetConversationByRemote(quoteDiffAccountID, "signal:"+peer) + if err != nil { + t.Fatalf("GetConversationByRemote: %v", err) + } + + clock := &reactionTestClock{now: start} + service := newSignalReactionService(t, v2Store, emptyLegacy, clock) + common := func(key string) messaging.CommonCommand { + return messaging.CommonCommand{ + AccountID: quoteDiffAccountID, ConversationID: v2Conversation.ConversationID, IdempotencyKey: key, + } + } + sendCommand := common("own-pending-send") + sendCommand.NotBefore = start.Add(time.Hour) + own, err := service.SendText(context.Background(), messaging.SendTextCommand{ + CommonCommand: sendCommand, + Body: "scheduled own message", + }) + if err != nil { + t.Fatalf("SendText(scheduled): %v", err) + } + reaction, err := service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: common("react-to-own-pending"), + TargetMessageID: own.LocalMessageID, + Emoji: "👍", + }) + if err != nil { + t.Fatalf("SendReaction(own pending): %v", err) + } + + if processed, err := service.DispatchDue(context.Background(), 4); err != nil || processed != 1 { + t.Fatalf("DispatchDue(reaction only) = %d, %v; want 1, nil", processed, err) + } + if calls := recorder.reactionCalls(); len(calls) != 0 { + t.Fatalf("reaction to a send with no Signal timestamp ran signal-cli: %q", calls) + } + delivery, err := service.Get(context.Background(), reaction.OutboxID) + if err != nil || delivery.State != messaging.OutboxNotDispatched || delivery.ErrorClass != string(bridge.FailureTransient) { + t.Fatalf("reaction delivery = %+v, %v; want not dispatched and retryable", delivery, err) + } + + recorder.scriptSend(1_760_003_600_555) + clock.advance(time.Hour + time.Second) + for attempt := 0; attempt < 4; attempt++ { + if _, err := service.DispatchDue(context.Background(), 4); err != nil { + t.Fatalf("DispatchDue(attempt %d): %v", attempt, err) + } + if delivery, err = service.Get(context.Background(), reaction.OutboxID); err != nil { + t.Fatalf("Get(reaction): %v", err) + } + if delivery.State == messaging.OutboxConfirmed { + break + } + clock.advance(10 * time.Second) + } + if sent, err := service.Get(context.Background(), own.OutboxID); err != nil || + sent.State != messaging.OutboxConfirmed || sent.RemoteMessageID != "1760003600555" { + t.Fatalf("own send delivery = %+v, %v; want confirmed at signal-cli's timestamp", sent, err) + } + if delivery.State != messaging.OutboxConfirmed { + t.Fatalf("reaction delivery after the send confirmed = %+v, want confirmed", delivery) + } + want := []string{ + "-a", quoteDiffAccount, + "sendReaction", + "-e", "👍", + "-a", quoteDiffAccount, + "-t", "1760003600555", + peer, + } + if calls := recorder.reactionCalls(); len(calls) != 1 || !slices.Equal(calls[0], want) { + t.Fatalf("signal-cli sendReaction calls = %q, want one %q", calls, want) + } +} + +// reactionTestClock is a messaging.Clock whose time moves only when told. +type reactionTestClock struct { + mu sync.Mutex + now time.Time +} + +func (c *reactionTestClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *reactionTestClock) advance(delay time.Duration) { + c.mu.Lock() + c.now = c.now.Add(delay) + c.mu.Unlock() +} + +func (c *reactionTestClock) NewTimer(delay time.Duration) messaging.Timer { + return messaging.SystemClock{}.NewTimer(delay) +} + +// TestV2ReactionRefusesATargetItCannotName covers stored messages whose +// author or Signal sent timestamp v2 cannot vouch for. Each would otherwise +// produce a well-formed signal-cli command naming a message that does not +// exist (the stored time, or this account as the author of someone else's +// message). The reaction must stop before signal-cli, stay retryable, and +// record why. +func TestV2ReactionRefusesATargetItCannotName(t *testing.T) { + const peer = "+15551234567" + const conversation = "signal:" + peer + const createdAt = int64(1_700_000_000_500) + const noTimestamp = "send_reaction: transient: signal reaction target timestamp is unavailable" + const noAuthor = "send_reaction: transient: signal reaction target author is unavailable" + tests := []struct { + name string + message sqlite.Message + fromPeer bool // stored with the peer's sender identity + wantDetail string + }{ + { + // migration/transform.go imports a legacy scheduled message caught + // in "sending" with a derived request ID, its creation time, and + // no outbox row. + name: "migrated scheduled send left in sending, with no outbox row", + message: sqlite.Message{ + RemoteMessageID: v2keys.DeriveID("transport_request", quoteDiffAccountID, "legacy-scheduled-1"), + Direction: sqlite.MessageDirectionOutgoing, + OccurredAtMS: createdAt, + }, + wantDetail: noTimestamp, + }, + { + // The legacy SendText stamped its row with the wall clock after + // signal-cli returned, so a migrated "local:" row's time may not + // be the timestamp Signal knows the message by. + name: "migrated own message under a local alias", + message: sqlite.Message{ + RemoteMessageID: v2keys.SignalLocalAlias(conversation, createdAt), + Direction: sqlite.MessageDirectionOutgoing, + OccurredAtMS: createdAt, + }, + wantDetail: noTimestamp, + }, + { + // The legacy receive handler stored a group message with no source + // under an empty sender; migration keeps it senderless. + name: "migrated incoming message with no sender", + message: sqlite.Message{ + RemoteMessageID: v2keys.SignalIncomingSourceID(conversation, "", createdAt), + Direction: sqlite.MessageDirectionIncoming, + OccurredAtMS: createdAt, + }, + wantDetail: noAuthor, + }, + { + name: "migrated incoming message with no sender and a decimal ID", + message: sqlite.Message{ + RemoteMessageID: "1700000000500", + Direction: sqlite.MessageDirectionIncoming, + OccurredAtMS: createdAt, + }, + wantDetail: noAuthor, + }, + { + // The Signal Desktop importer stores a row with no sent time under + // the time it was received and marks its source ID, which the + // migration keeps as the remote ID. + name: "imported Signal Desktop row stored under its received time", + message: sqlite.Message{ + RemoteMessageID: v2keys.SignalReceivedSourceID(conversation, peer, createdAt), + Direction: sqlite.MessageDirectionIncoming, + OccurredAtMS: createdAt, + }, + fromPeer: true, + wantDetail: noTimestamp, + }, + { + name: "incoming message from a known sender under a decimal ID", + message: sqlite.Message{ + RemoteMessageID: "1700000000500", + Direction: sqlite.MessageDirectionIncoming, + OccurredAtMS: createdAt, + }, + fromPeer: true, + wantDetail: noTimestamp, + }, + { + // v2wire.MirrorReplyTarget writes an incoming message under its + // legacy ID with no sender identity. + name: "mirrored incoming message with no sender", + message: sqlite.Message{ + RemoteMessageID: "signal:1700000000500", + Direction: sqlite.MessageDirectionIncoming, + OccurredAtMS: createdAt, + }, + wantDetail: noAuthor, + }, + } + for index, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + recorder := installSignalCLIRecorder(t) + dir := t.TempDir() + emptyLegacy := openQuoteDiffLegacy(t, filepath.Join(dir, "legacy.sqlite3")) + v2Store, messages, sink, worker := openQuoteDiffV2(t, filepath.Join(dir, "v2.sqlite3")) + line := []byte(fmt.Sprintf( + `{"account":%q,"envelope":{"source":%q,"sourceNumber":%q,"timestamp":%d,"dataMessage":{"timestamp":%d,"message":"hello"}}}`, + quoteDiffAccount, peer, peer, createdAt-60_000, createdAt-60_000, + )) + ingestSignalLine(t, sink, worker, messages, line, peer, time.UnixMilli(createdAt-60_000)) + v2Conversation, err := v2Store.GetConversationByRemote(quoteDiffAccountID, conversation) + if err != nil { + t.Fatalf("GetConversationByRemote: %v", err) + } + + message := tc.message + if tc.fromPeer { + ingested, err := messages.ListMessagesByConversation(context.Background(), v2Conversation.ConversationID, 0, "", 10) + if err != nil || len(ingested) != 1 || ingested[0].SenderIdentityID == nil { + t.Fatalf("ingested peer message = %+v, %v; want one with a sender identity", ingested, err) + } + message.SenderIdentityID = ingested[0].SenderIdentityID + } + message.MessageID = fmt.Sprintf("message-unnameable-%d", index) + message.ConversationID = v2Conversation.ConversationID + message.AccountID = quoteDiffAccountID + message.Body = "stored without a nameable Signal identity" + message.State = sqlite.MessageStateActive + if err := messages.ImportMessage(context.Background(), sqlite.MessageProjection{Message: message}); err != nil { + t.Fatalf("ImportMessage: %v", err) + } + + service := newSignalReactionService(t, v2Store, emptyLegacy, messaging.SystemClock{}) + submission, err := service.SendReaction(context.Background(), messaging.SendReactionCommand{ + CommonCommand: messaging.CommonCommand{ + AccountID: quoteDiffAccountID, + ConversationID: v2Conversation.ConversationID, + IdempotencyKey: fmt.Sprintf("react-unnameable-%d", index), + }, + TargetMessageID: message.MessageID, + Emoji: "👍", + }) + if err != nil { + t.Fatalf("SendReaction(): %v", err) + } + if processed, err := service.DispatchDue(context.Background(), 1); err != nil || processed != 1 { + t.Fatalf("DispatchDue() = %d, %v; want 1, nil", processed, err) + } + if calls := recorder.reactionCalls(); len(calls) != 0 { + t.Fatalf("signal-cli ran %q for a target v2 cannot name", calls) + } + delivery, err := service.Get(context.Background(), submission.OutboxID) + if err != nil || delivery.State != messaging.OutboxNotDispatched || + delivery.ErrorClass != string(bridge.FailureTransient) { + t.Fatalf("reaction delivery = %+v, %v; want not dispatched and retryable", delivery, err) + } + outbox, err := sqlite.NewOutboxRepository(v2Store, time.Now) + if err != nil { + t.Fatalf("NewOutboxRepository(): %v", err) + } + row, err := outbox.FindByID(context.Background(), submission.OutboxID) + if err != nil || row.ErrorDetail == nil || *row.ErrorDetail != tc.wantDetail { + t.Fatalf("reaction row = %+v, %v; want error detail %q", row, err, tc.wantDetail) + } + }) + } +} diff --git a/internal/signallive/reaction_target_test.go b/internal/signallive/reaction_target_test.go new file mode 100644 index 00000000..d7654892 --- /dev/null +++ b/internal/signallive/reaction_target_test.go @@ -0,0 +1,431 @@ +package signallive + +import ( + "context" + "errors" + "math/rand" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/rs/zerolog" + + "github.com/maxghenis/openmessage/internal/v2keys" +) + +const ( + reactionTestAccount = "+15551230000" + reactionTestSender = "+15551234567" + reactionTestACI = "9f4b50e3-ebf2-413c-a856-161756a6161a" + // An outbox transport request ID: 128 random bits in hex + // (messaging.CryptoIDSource), the remote ID of an unconfirmed send. + reactionTestRequestID = "0f1e2d3c4b5a69788796a5b4c3d2e1f0" + reactionTestLocalID = "local:c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735" + errReactionTimestamp = "signal reaction target timestamp is unavailable" + errReactionAuthor = "signal reaction target author is unavailable" +) + +func TestReactionTargetArgsRules(t *testing.T) { + incoming := v2keys.SignalIncomingSourceID("signal:"+reactionTestSender, reactionTestSender, 1700000000123) + tests := []struct { + name string + target ReactionTarget + account string + want []string + wantErr string + }{ + { + name: "incoming message: SHA-1 remote ID, timestamp from the occurred time", + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + want: reactionArgs(reactionTestSender, "1700000000123"), + }, + { + name: "incoming from an unresolved ACI keeps the ACI as author", + target: ReactionTarget{RemoteID: incoming, AuthorID: " " + reactionTestACI + " ", SentAt: time.UnixMilli(1700000000124)}, + want: reactionArgs(reactionTestACI, "1700000000124"), + }, + { + name: "incoming message stored under this account's own address is still incoming", + target: ReactionTarget{RemoteID: " " + incoming + " ", AuthorID: reactionTestAccount, SentAt: time.UnixMilli(1700000000123)}, + want: reactionArgs(reactionTestAccount, "1700000000123"), + }, + { + name: "incoming Signal Desktop row stored under its received time", + target: ReactionTarget{ + RemoteID: v2keys.SignalReceivedSourceID("signal:"+reactionTestSender, reactionTestSender, 1700000000500), + AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000500), + }, + wantErr: errReactionTimestamp, + }, + { + name: "incoming under a decimal remote ID, a form no Signal receiver writes", + target: ReactionTarget{RemoteID: "1700000000999", AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionTimestamp, + }, + { + name: "mirrored incoming message under its legacy ID", + target: ReactionTarget{RemoteID: "signal:" + incoming, AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionTimestamp, + }, + { + name: "incoming under an upper-case hash is not the receiver's form", + target: ReactionTarget{RemoteID: strings.ToUpper(incoming), AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionTimestamp, + }, + { + name: "own outbox send: the confirmed transport timestamp, not the submit time", + target: ReactionTarget{RemoteID: "1700000000555", Outgoing: true, SentAt: time.UnixMilli(1700000000000)}, + want: reactionArgs(reactionTestAccount, "1700000000555"), + }, + { + name: "own send from the phone: remote ID and occurred time agree", + target: ReactionTarget{RemoteID: " 1700000000777 ", Outgoing: true, SentAt: time.UnixMilli(1700000000777)}, + want: reactionArgs(reactionTestAccount, "1700000000777"), + }, + { + name: "own send is this account's whatever sender the store recorded", + target: ReactionTarget{RemoteID: "1700000000556", AuthorID: reactionTestACI, Outgoing: true}, + want: reactionArgs(reactionTestAccount, "1700000000556"), + }, + { + name: "own send still on its outbox request ID", + target: ReactionTarget{RemoteID: reactionTestRequestID, Outgoing: true, SentAt: time.UnixMilli(1700000000000)}, + wantErr: errReactionTimestamp, + }, + { + name: "own migrated message under a local alias", + target: ReactionTarget{RemoteID: reactionTestLocalID, Outgoing: true, SentAt: time.UnixMilli(1700000000321)}, + wantErr: errReactionTimestamp, + }, + { + name: "own migrated placeholder for an edit", + target: ReactionTarget{RemoteID: "missing-edit:c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735", Outgoing: true, SentAt: time.UnixMilli(1700000000322)}, + wantErr: errReactionTimestamp, + }, + { + name: "own decimal remote ID past int64", + target: ReactionTarget{RemoteID: "99999999999999999999", Outgoing: true, SentAt: time.UnixMilli(1700000000322)}, + wantErr: errReactionTimestamp, + }, + { + name: "own zero remote ID", + target: ReactionTarget{RemoteID: "0", Outgoing: true, SentAt: time.UnixMilli(1700000000323)}, + wantErr: errReactionTimestamp, + }, + { + name: "incoming message with no stored sender", + target: ReactionTarget{RemoteID: incoming, SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionAuthor, + }, + { + name: "incoming message with no stored sender and a decimal remote ID", + target: ReactionTarget{RemoteID: "1700000000123", SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionAuthor, + }, + { + name: "mirrored incoming message with no stored sender", + target: ReactionTarget{RemoteID: "signal:1700000000123", SentAt: time.UnixMilli(1700000000123)}, + wantErr: errReactionAuthor, + }, + { + name: "incoming message with no occurred time", + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender}, + wantErr: errReactionTimestamp, + }, + { + name: "an epoch occurred time is not a timestamp", + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender, SentAt: time.UnixMilli(0)}, + wantErr: errReactionTimestamp, + }, + { + name: "a negative occurred time is not a timestamp", + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender, SentAt: time.UnixMilli(-5)}, + wantErr: errReactionTimestamp, + }, + { + name: "own send with no account to name", + target: ReactionTarget{RemoteID: "1700000000555", Outgoing: true}, + account: " ", + wantErr: errReactionAuthor, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + account := reactionTestAccount + if tc.account != "" { + account = tc.account + } + got, err := ReactionTargetArgs(tc.target, account) + if tc.wantErr != "" { + if err == nil || err.Error() != tc.wantErr || got != nil { + t.Fatalf("ReactionTargetArgs(%+v) = %q, %v; want error %q", tc.target, got, err, tc.wantErr) + } + return + } + if err != nil || !slices.Equal(got, tc.want) { + t.Fatalf("ReactionTargetArgs(%+v) = %q, %v; want %q", tc.target, got, err, tc.want) + } + }) + } +} + +// TestReactionTargetArgsInvariants checks, over seeded random targets, against +// an independently written oracle: +// +// - Shape: success is exactly "-a -t ". +// - Own message: the author is the account, and the timestamp is the +// RemoteID when it is a positive int64 decimal. Any other RemoteID (a +// request ID, a "local:" or "missing-edit:" alias, a SHA-1) fails as +// unavailable whatever SentAt holds. +// - Incoming message: the author is the trimmed AuthorID and the timestamp +// is SentAt, and only when RemoteID is 40 lowercase hex digits, the form a +// Signal receiver gives a message it keyed by its sent timestamp. No +// AuthorID fails as author unavailable; any other RemoteID (a decimal, a +// "received:" marker, a legacy "signal:" ID) or a SentAt that is zero or +// not positive fails as timestamp unavailable. +// - So -t is never a RemoteID that is not a decimal of an own message, never +// the stored time of a message whose ID does not vouch for it, and -a is +// never the account for an incoming message that names no sender. +// - Agreement with quotes: wherever a reaction names a message, QuoteArgs +// names the same author and timestamp for it. +// - Determinism. +func TestReactionTargetArgsInvariants(t *testing.T) { + random := rand.New(rand.NewSource(20261010)) + authors := []string{"", " ", reactionTestAccount, " " + reactionTestAccount, reactionTestSender, reactionTestACI} + remoteIDs := []string{ + "", "0", "-17", "+17", "1700000000555", " 1700000000556 ", "99999999999999999999", + "9223372036854775807", "9223372036854775808", "signal:1700000000557", + "f48818f15483f503bb133d92360ca8e2fbd8287e", " 36c1e3741b6432157a0af5fe136a306fd941ee21 ", + "0000000000000000000000000000000000000000", "c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735", + "F48818F15483F503BB133D92360CA8E2FBD8287E", "f48818f15483f503bb133d92360ca8e2fbd8287", "f48818f15483f503bb133d92360ca8e2fbd8287e0", + "received:f48818f15483f503bb133d92360ca8e2fbd8287e", "signal:f48818f15483f503bb133d92360ca8e2fbd8287e", + reactionTestLocalID, reactionTestRequestID, + "missing-edit:c88a3b1e5f4ec4fe43c3b85b7664cbe45b711735", "17000000005x5", + } + identity := func(value string) string { return value } + ownSuccesses, incomingSuccesses, authorFailures, timestampFailures := 0, 0, 0, 0 + for iteration := 0; iteration < 5000; iteration++ { + target := ReactionTarget{ + RemoteID: remoteIDs[random.Intn(len(remoteIDs))], + AuthorID: authors[random.Intn(len(authors))], + Outgoing: random.Intn(2) == 0, + } + 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 := ReactionTargetArgs(target, reactionTestAccount) + again, againErr := ReactionTargetArgs(target, reactionTestAccount) + if !slices.Equal(got, again) || errorText(err) != errorText(againErr) { + t.Fatalf("iteration %d: ReactionTargetArgs(%+v) is not deterministic", iteration, target) + } + + var wantAuthor string + var wantTimestamp int64 + if target.Outgoing { + wantAuthor = reactionTestAccount + if trimmed := strings.TrimSpace(target.RemoteID); trimmed != "" && strings.Trim(trimmed, "0123456789") == "" { + if parsed, parseErr := strconv.ParseInt(trimmed, 10, 64); parseErr == nil { + wantTimestamp = parsed + } + } + } else { + wantAuthor = strings.TrimSpace(target.AuthorID) + trimmed := strings.TrimSpace(target.RemoteID) + receiverForm := len(trimmed) == 40 && strings.Trim(trimmed, "0123456789abcdef") == "" + if receiverForm && !target.SentAt.IsZero() { + wantTimestamp = target.SentAt.UnixMilli() + } + } + switch { + case wantAuthor == "": + authorFailures++ + if err == nil || err.Error() != errReactionAuthor || got != nil { + t.Fatalf("iteration %d: ReactionTargetArgs(%+v) = %q, %v; want unavailable author", iteration, target, got, err) + } + continue + case wantTimestamp <= 0: + timestampFailures++ + if err == nil || err.Error() != errReactionTimestamp || got != nil { + t.Fatalf("iteration %d: ReactionTargetArgs(%+v) = %q, %v; want unavailable timestamp", iteration, target, got, err) + } + continue + } + if target.Outgoing { + ownSuccesses++ + } else { + incomingSuccesses++ + } + want := reactionArgs(wantAuthor, strconv.FormatInt(wantTimestamp, 10)) + if err != nil || !slices.Equal(got, want) { + t.Fatalf("iteration %d: ReactionTargetArgs(%+v) = %q, %v; want %q", iteration, target, got, err, want) + } + + // The quote of the same stored message. QuoteArgs has no direction: it + // reads an empty author as this account, which is how an own message + // reaches it. An incoming message stored under this account's own + // address is the one shape it reads differently, so it is left out. + quoted := ReplyTarget{RemoteID: target.RemoteID, AuthorID: target.AuthorID, SentAt: target.SentAt, Text: "quoted"} + if target.Outgoing { + quoted.AuthorID = "" + } else if wantAuthor == reactionTestAccount { + continue + } + quote, quoteErr := QuoteArgs(quoted, reactionTestAccount, identity) + if quoteErr != nil || len(quote) != 6 || quote[1] != got[3] || quote[3] != got[1] { + t.Fatalf("iteration %d: QuoteArgs(%+v) = %q, %v; the reaction targets %q", iteration, quoted, quote, quoteErr, got) + } + } + if ownSuccesses < 150 || incomingSuccesses < 150 || authorFailures < 200 || timestampFailures < 500 { + t.Fatalf("generator covered %d own and %d incoming successes, %d author failures, %d timestamp failures", + ownSuccesses, incomingSuccesses, authorFailures, timestampFailures) + } +} + +// TestSendReactionRequestTargetsTheDescribedMessage checks the signal-cli argv +// SendReactionRequest runs for each kind of target the v2 dispatcher hands it, +// and that a target it cannot name fails before signal-cli runs, with an +// error the adapter reports as not dispatched. +func TestSendReactionRequestTargetsTheDescribedMessage(t *testing.T) { + incoming := v2keys.SignalIncomingSourceID("signal:"+reactionTestSender, reactionTestSender, 1700000000123) + tests := []struct { + name string + conversation string + target ReactionTarget + action string + want []string + wantErr string + }{ + { + name: "incoming message in a 1:1 conversation", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + action: "add", + want: []string{ + "-a", reactionTestAccount, "sendReaction", "-e", "👍", + "-a", reactionTestSender, "-t", "1700000000123", reactionTestSender, + }, + }, + { + name: "incoming message from an ACI in a group, removed", + conversation: "signal-group:test-group", + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestACI, SentAt: time.UnixMilli(1700000000124)}, + action: "remove", + want: []string{ + "-a", reactionTestAccount, "sendReaction", "-e", "👍", + "-a", reactionTestACI, "-t", "1700000000124", "-r", "--group-id", "test-group", + }, + }, + { + name: "own confirmed outbox send", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{RemoteID: "1700000000555", Outgoing: true, SentAt: time.UnixMilli(1700000000000)}, + action: "switch", + want: []string{ + "-a", reactionTestAccount, "sendReaction", "-e", "👍", + "-a", reactionTestAccount, "-t", "1700000000555", reactionTestSender, + }, + }, + { + name: "own send still on its outbox request ID", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{RemoteID: reactionTestRequestID, Outgoing: true, SentAt: time.UnixMilli(1700000000000)}, + action: "add", + wantErr: errReactionTimestamp, + }, + { + name: "own migrated message under a local alias", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{RemoteID: reactionTestLocalID, Outgoing: true, SentAt: time.UnixMilli(1700000000321)}, + action: "add", + wantErr: errReactionTimestamp, + }, + { + name: "incoming message with no stored sender", + conversation: "signal-group:test-group", + target: ReactionTarget{RemoteID: "1700000000123", SentAt: time.UnixMilli(1700000000123)}, + action: "add", + wantErr: errReactionAuthor, + }, + { + name: "incoming message with no occurred time", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{RemoteID: incoming, AuthorID: reactionTestSender}, + action: "add", + wantErr: errReactionTimestamp, + }, + { + name: "incoming Signal Desktop row stored under its received time", + conversation: "signal:" + reactionTestSender, + target: ReactionTarget{ + RemoteID: v2keys.SignalReceivedSourceID("signal:"+reactionTestSender, reactionTestSender, 1700000000500), + AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000500), + }, + action: "add", + wantErr: errReactionTimestamp, + }, + } + + originalRun := runSignalCLI + t.Cleanup(func() { runSignalCLI = originalRun }) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + bridge := &Bridge{ + account: reactionTestAccount, + connected: true, + configDir: t.TempDir(), + logger: zerolog.Nop(), + } + var calls [][]string + runSignalCLI = func(_ context.Context, _ string, args ...string) ([]byte, error) { + calls = append(calls, slices.Clone(args)) + return []byte("ok"), nil + } + + err := bridge.SendReactionRequest(tc.conversation, tc.target, "👍", tc.action) + if tc.wantErr != "" { + if err == nil || err.Error() != tc.wantErr || IsCommandError(err) || IsSendNotDispatchedError(err) { + t.Fatalf("SendReactionRequest() = %v (%T); want plain pre-call error %q", err, err, tc.wantErr) + } + if len(calls) != 0 { + t.Fatalf("signal-cli calls = %q, want none for a target it cannot name", calls) + } + return + } + if err != nil { + t.Fatalf("SendReactionRequest(): %v", err) + } + if len(calls) != 1 || !slices.Equal(calls[0], tc.want) { + t.Fatalf("signal-cli calls = %q, want one %q", calls, tc.want) + } + }) + } +} + +func TestSendReactionRequestStillRequiresATargetRemoteID(t *testing.T) { + bridge := &Bridge{account: reactionTestAccount, connected: true, configDir: t.TempDir(), logger: zerolog.Nop()} + originalRun := runSignalCLI + t.Cleanup(func() { runSignalCLI = originalRun }) + runSignalCLI = func(context.Context, string, ...string) ([]byte, error) { + return nil, errors.New("signal-cli must not run") + } + err := bridge.SendReactionRequest( + "signal:"+reactionTestSender, + ReactionTarget{RemoteID: " ", AuthorID: reactionTestSender, SentAt: time.UnixMilli(1700000000123)}, + "👍", + "add", + ) + if err == nil || err.Error() != "signal target message is required" { + t.Fatalf("SendReactionRequest(no remote ID) = %v, want the required-target error", err) + } +} + +func reactionArgs(author, timestamp string) []string { + return []string{"-a", author, "-t", timestamp} +} diff --git a/internal/signallive/reply_quote.go b/internal/signallive/reply_quote.go index da46f94a..874cc6cf 100644 --- a/internal/signallive/reply_quote.go +++ b/internal/signallive/reply_quote.go @@ -13,10 +13,13 @@ import ( // 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). +// was (signalQuoteArgs). Outgoing reports that this account sent the message; +// QuoteArgs does not read it and still takes an empty AuthorID as this +// account. type ReplyTarget struct { RemoteID string AuthorID string + Outgoing bool SentAt time.Time Text string HasAttachment bool diff --git a/internal/v2keys/derive.go b/internal/v2keys/derive.go index 2bbc620d..c5145fb6 100644 --- a/internal/v2keys/derive.go +++ b/internal/v2keys/derive.go @@ -91,6 +91,31 @@ func SignalIncomingSourceID(conversationID, source string, timestamp int64) stri return hex.EncodeToString(sum[:]) } +// IsSignalIncomingSourceID reports whether id has the form +// SignalIncomingSourceID returns: 40 lowercase hexadecimal digits. An incoming +// Signal message stored under such an ID was keyed by a receiver from its +// sender and sent timestamp. +func IsSignalIncomingSourceID(id string) bool { + if len(id) != sha1.Size*2 { + return false + } + for _, r := range id { + if (r < '0' || r > '9') && (r < 'a' || r > 'f') { + return false + } + } + return true +} + +// SignalReceivedSourceID returns the source ID of an incoming Signal message +// that was stored with no sent timestamp, only the time it was received (a +// Signal Desktop row without one). Signal names a message by its sender and +// sent timestamp, so this message has no Signal identity. The prefix keeps it +// from passing for one keyed by SignalIncomingSourceID. +func SignalReceivedSourceID(conversationID, source string, receivedAt int64) string { + return "received:" + SignalIncomingSourceID(conversationID, source, receivedAt) +} + // SignalLocalAlias returns the migration remote-message key for a fabricated // local Signal message. func SignalLocalAlias(conversationID string, timestamp int64) string { diff --git a/internal/v2keys/derive_test.go b/internal/v2keys/derive_test.go index 8bd5f69b..75b0fd24 100644 --- a/internal/v2keys/derive_test.go +++ b/internal/v2keys/derive_test.go @@ -143,3 +143,50 @@ func TestSignalLocalAlias(t *testing.T) { t.Fatalf("SignalLocalAlias() = %q, want %q", got, want) } } + +func TestSignalReceivedSourceIDIsNotAnIncomingSourceID(t *testing.T) { + t.Parallel() + + const conversation, source, when = "signal:+16505550100", "+16505550100", int64(1_700_000_001_000) + incoming := SignalIncomingSourceID(conversation, source, when) + if !IsSignalIncomingSourceID(incoming) { + t.Fatalf("IsSignalIncomingSourceID(%q) = false for SignalIncomingSourceID's own output", incoming) + } + received := SignalReceivedSourceID(conversation, source, when) + if want := "received:" + incoming; received != want { + t.Fatalf("SignalReceivedSourceID() = %q, want %q", received, want) + } + // The same message stored under a received time must never read as one + // keyed by its sent timestamp. + if IsSignalIncomingSourceID(received) { + t.Fatalf("IsSignalIncomingSourceID(%q) = true for a received-time ID", received) + } +} + +func TestIsSignalIncomingSourceID(t *testing.T) { + t.Parallel() + + sha := SignalIncomingSourceID("signal:+16505550100", "+16505550100", 1_700_000_001_000) + tests := []struct { + id string + want bool + }{ + {id: sha, want: true}, + {id: "0000000000000000000000000000000000000000", want: true}, + {id: "", want: false}, + {id: sha[:39], want: false}, + {id: sha + "0", want: false}, + {id: " " + sha[1:], want: false}, + {id: "A" + sha[1:], want: false}, + {id: "g" + sha[1:], want: false}, + {id: "1700000001000", want: false}, + {id: "signal:1700000001000", want: false}, + {id: SignalLocalAlias("signal:+16505550100", 1_700_000_001_000), want: false}, + {id: DeriveID("transport_request", "signal-primary", "scheduled-1"), want: false}, + } + for _, test := range tests { + if got := IsSignalIncomingSourceID(test.id); got != test.want { + t.Errorf("IsSignalIncomingSourceID(%q) = %v, want %v", test.id, got, test.want) + } + } +}