package message import ( "context" "database/sql" "testing" "time" "github.com/stretchr/testify/require" "github.com/tidbcloud/AgentTeam/internal/coredb" ) func TestScopeRewindSuffixRemovesReadProjectionAndSearchVisibility(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) const ( workspaceID = "wsp_scope_rewind_read_projection" ownerUserID = "usr_scope_rewind_read_projection_owner" ownerMemberID = "mem_scope_rewind_read_projection_owner" agentMemberID = "mem_scope_rewind_read_projection_agent" scopeID = "scp_scope_rewind_read_projection" keptMessageID = "msg_scope_rewind_read_projection_keep" rewoundMessageID = "msg_scope_rewind_read_projection_rewound" survivorMessageID = "msg_scope_rewind_read_projection_survivor" ) insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-rewind-read-projection@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) insertMessagingAgentMember(t, ctx, db, agentMemberID, workspaceID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: scopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-rewind-read-projection", LatestSeq: 2, LatestGlobalSeq: 20, LatestMessageAt: timePtr(now.Add(2 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", LastReadSeq: 2, LastReadGlobalSeq: 20, OwnerContextDismissedThroughGlobalSeq: 20, EffectiveFromSeq: 1, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: agentMemberID, Visibility: "public", LastReadSeq: 0, LastReadGlobalSeq: 0, EffectiveFromSeq: 1, CreatedAt: now, UpdatedAt: now, })) keptMessage := ScopeMessageRecord{ ID: keptMessageID, WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 1, WorkspaceGlobalSeq: 10, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "rewind visibility keep needle", Metadata: []byte(`{}`), CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), } rewoundMessage := ScopeMessageRecord{ ID: rewoundMessageID, WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 2, WorkspaceGlobalSeq: 20, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "rewind visibility rewound needle", Metadata: []byte(`{}`), CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), } for _, record := range []ScopeMessageRecord{keptMessage, rewoundMessage} { require.NoError(t, repos.Messages.Insert(ctx, record)) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, })) } require.NoError(t, repos.Sources.Insert(ctx, ScopeMessageSourceRecord{ TargetScopeID: scopeID, MessageID: keptMessageID, Ordinal: 0, SourceScopeID: scopeID, SourceSeq: 2, SourceMessageID: rewoundMessageID, SourceKind: "quote", SourceDedupeKey: "source:scope-rewind-read-projection-rewound-source", CreatedAt: now, })) require.NoError(t, repos.Sources.Insert(ctx, ScopeMessageSourceRecord{ TargetScopeID: scopeID, MessageID: rewoundMessageID, Ordinal: 0, SourceScopeID: scopeID, SourceSeq: 1, SourceMessageID: keptMessageID, SourceKind: "quote", SourceDedupeKey: "source:scope-rewind-read-projection-rewound-target", CreatedAt: now, })) mustExecMessaging(t, ctx, db, ` INSERT INTO attachments ( id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, storage_driver, payload, created_at ) VALUES ( 'att_scope_rewind_read_projection_rewound', $1, $2, $3, $4, 'rewound.txt', 'text/plain', 7, 'inline', 'rewound', $5 ) `, workspaceID, scopeID, ownerMemberID, rewoundMessageID, now) require.NoError(t, repos.ShareCards.Insert(ctx, ScopeShareCardRecord{ ID: "shr_scope_rewind_read_projection_rewound", WorkspaceID: workspaceID, ScopeID: scopeID, MessageID: rewoundMessageID, SourceScopeID: scopeID, SourceMessageID: keptMessageID, TargetScopeID: scopeID, CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) mustExecMessaging(t, ctx, db, ` INSERT INTO im_stream_events ( workspace_id, member_id, event_seq, event_id, type, scope_id, conversation_id, entity_id, entity_seq, payload_json, patches_json, cursor_generation, created_at, expires_at ) VALUES ( $1, $2, 1, 'evt_scope_rewind_read_projection_rewound_created', 'message.created', $3, $3, $4, $5, $6, NULL, '', $7, NULL ) `, workspaceID, ownerMemberID, scopeID, rewoundMessageID, rewoundMessage.Seq, `{ "content": "rewind visibility rewound needle", "mention": {"preview": "rewind visibility rewound needle"}, "attachments": [{"id": "att_scope_rewind_read_projection_rewound", "filename": "rewound.txt"}], "message": { "id": "`+rewoundMessageID+`", "content": "rewind visibility rewound needle", "metadata": {"mentions": [{"id": "agt"}], "agentDelivery": {"status": "delivered"}}, "resolvedMentions": [{"id": "agt"}], "attachments": [{"id": "att_scope_rewind_read_projection_rewound", "filename": "rewound.txt"}], "source": {"snippet": "rewind visibility rewound needle"}, "deliverySummaryPreview": {"targetedAgentCount": 1} } }`, now) mustExecMessaging(t, ctx, db, ` INSERT INTO im_conversation_index ( workspace_id, member_id, conversation_id, scope_id, kind, sort_at, last_event_seq, pinned_at, archived_at, projection_json, created_at, updated_at ) VALUES ( $1, $2, $3, $3, 'channel', $4, 1, NULL, NULL, $5, $4, $4 ) `, workspaceID, ownerMemberID, scopeID, now, `{ "content": "rewind visibility rewound needle", "attachments": [{"id": "att_scope_rewind_read_projection_rewound", "filename": "rewound.txt"}], "lastMessage": { "id": "`+rewoundMessageID+`", "content": "rewind visibility rewound needle", "metadata": {"mentions": [{"id": "agt"}], "agentDelivery": {"status": "delivered"}}, "resolvedMentions": [{"id": "agt"}], "attachments": [{"id": "att_scope_rewind_read_projection_rewound", "filename": "rewound.txt"}], "source": {"snippet": "rewind visibility rewound needle"}, "deliverySummaryPreview": {"targetedAgentCount": 1} } }`) api := NewScopeAPIService(New(db, nil, 0, Options{})) beforeHistory, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Limit: 10, AdvanceReadState: true, }) require.NoError(t, err) require.Equal(t, []string{rewoundMessageID, keptMessageID}, historyMessageIDs(beforeHistory.Messages)) beforeReadCheck, err := api.CanReadMessage(ctx, ScopeAPIReadCheckRequest{ WorkspaceID: workspaceID, MessageID: rewoundMessageID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, RequestKind: ScopeVisibilityRequestAttachment, }) require.NoError(t, err) require.True(t, beforeReadCheck.Allowed) beforeSearch, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 10, CandidateBatchLimit: 10, MaxCandidateBatches: 1, }, 10, func(context.Context, MessageSearchCursor, int) ([]ScopeMessageVisibilityCandidate, error) { return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(rewoundMessage, ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", EffectiveFromSeq: 1}), scopeSearchCandidate(keptMessage, ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", EffectiveFromSeq: 1}), }, nil }) require.NoError(t, err) require.Equal(t, []string{rewoundMessageID, keptMessageID}, visibleMessageRefIDs(beforeSearch.Messages)) beforeDiagnostics, err := New(db, nil, 0, Options{}).GetMessageDeliveryDiagnosticsForUser(ctx, ownerUserID, workspaceID, rewoundMessageID) require.NoError(t, err) require.Equal(t, rewoundMessageID, beforeDiagnostics.Message.ID) beforeOwnerQuoteSources, err := LoadOwnerQuoteSourceMessages(ctx, db, workspaceID, []string{rewoundMessageID}) require.NoError(t, err) require.Equal(t, []string{rewoundMessageID}, ownerQuoteSourceMessageIDsForRewindTest(beforeOwnerQuoteSources)) require.Equal(t, []string{rewoundMessageID}, syncMessageIDsForRewindTest(t, ctx, db, workspaceID, MessageSyncDirtyScopeCandidate{ ScopeID: scopeID, FrontierSeq: 1, LatestMessageSeq: 2, })) require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_share_cards", "message_id = $1", rewoundMessageID)) result, err := New(db, nil, 0, Options{}).RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, FromSeq: 2, RewoundAt: now.Add(time.Hour), }) require.NoError(t, err) require.Equal(t, []string{rewoundMessageID}, result.RewoundMessageIDs) require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_messages", "id = $1", rewoundMessageID)) var rewoundContent string var rewoundMetadata []byte require.NoError(t, db.QueryRow(ctx, `SELECT content, metadata FROM scope_messages WHERE id = $1`, rewoundMessageID).Scan(&rewoundContent, &rewoundMetadata)) require.Empty(t, rewoundContent) decodedRewoundMetadata, err := DecodeMessageMetadata(rewoundMetadata) require.NoError(t, err) require.True(t, MessageMetadataIsRewound(decodedRewoundMetadata)) require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_search_docs", "message_id = $1", rewoundMessageID)) require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_sources", "message_id = $1 OR source_message_id = $1", rewoundMessageID)) require.Equal(t, int64(0), countScopeRows(t, ctx, db, "attachments", "message_id = $1", rewoundMessageID)) require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_share_cards", "message_id = $1", rewoundMessageID)) var streamPayload, streamMessageContent, streamAttachmentJSON, streamRewoundStatus string var streamMention sql.NullString require.NoError(t, db.QueryRow(ctx, ` SELECT CAST(payload_json AS CHAR), COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$.message.content')), ''), COALESCE(CAST(JSON_EXTRACT(payload_json, '$.message.attachments') AS CHAR), ''), COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$.message.metadata.messageRecall.status')), ''), JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$.mention')) FROM im_stream_events WHERE workspace_id = $1 AND member_id = $2 AND event_id = 'evt_scope_rewind_read_projection_rewound_created' `, workspaceID, ownerMemberID).Scan(&streamPayload, &streamMessageContent, &streamAttachmentJSON, &streamRewoundStatus, &streamMention)) require.NotContains(t, streamPayload, "rewind visibility rewound needle") require.Empty(t, streamMessageContent) require.Equal(t, "[]", streamAttachmentJSON) require.Equal(t, MessageLifecycleStatusRewound, streamRewoundStatus) require.False(t, streamMention.Valid) var projectionJSON, projectionLastContent, projectionAttachmentJSON, projectionRewoundStatus string require.NoError(t, db.QueryRow(ctx, ` SELECT CAST(projection_json AS CHAR), COALESCE(JSON_UNQUOTE(JSON_EXTRACT(projection_json, '$.lastMessage.content')), ''), COALESCE(CAST(JSON_EXTRACT(projection_json, '$.lastMessage.attachments') AS CHAR), ''), COALESCE(JSON_UNQUOTE(JSON_EXTRACT(projection_json, '$.lastMessage.metadata.messageRecall.status')), '') FROM im_conversation_index WHERE workspace_id = $1 AND member_id = $2 AND conversation_id = $3 `, workspaceID, ownerMemberID, scopeID).Scan(&projectionJSON, &projectionLastContent, &projectionAttachmentJSON, &projectionRewoundStatus)) require.NotContains(t, projectionJSON, "rewind visibility rewound needle") require.Empty(t, projectionLastContent) require.Equal(t, "[]", projectionAttachmentJSON) require.Equal(t, MessageLifecycleStatusRewound, projectionRewoundStatus) var lastReadSeq, lastReadGlobalSeq, dismissedGlobalSeq int64 require.NoError(t, db.QueryRow(ctx, ` SELECT last_read_seq, last_read_global_seq, owner_context_dismissed_through_global_seq FROM scope_memberships WHERE scope_id = $1 AND member_id = $2 `, scopeID, ownerMemberID).Scan(&lastReadSeq, &lastReadGlobalSeq, &dismissedGlobalSeq)) require.Equal(t, int64(1), lastReadSeq) require.Equal(t, int64(10), lastReadGlobalSeq) require.Equal(t, int64(10), dismissedGlobalSeq) afterHistory, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Limit: 10, AdvanceReadState: true, }) require.NoError(t, err) require.Equal(t, []string{keptMessageID}, historyMessageIDs(afterHistory.Messages)) afterReadCheck, err := api.CanReadMessage(ctx, ScopeAPIReadCheckRequest{ WorkspaceID: workspaceID, MessageID: rewoundMessageID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, RequestKind: ScopeVisibilityRequestAttachment, }) require.NoError(t, err) require.False(t, afterReadCheck.Allowed) afterSearch, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 10, CandidateBatchLimit: 10, MaxCandidateBatches: 1, }, 10, func(ctx context.Context, _ MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { return rewindSearchCandidatesFromProjectedDocsForTest(t, ctx, db, ownerMemberID, keptMessage), nil }) require.NoError(t, err) require.Equal(t, []string{keptMessageID}, visibleMessageRefIDs(afterSearch.Messages)) _, err = New(db, nil, 0, Options{}).GetMessageDeliveryDiagnosticsForUser(ctx, ownerUserID, workspaceID, rewoundMessageID) require.ErrorIs(t, err, ErrNotFound) _, err = LoadOwnerQuoteSourceMessages(ctx, db, workspaceID, []string{rewoundMessageID}) require.ErrorContains(t, err, "source messages must belong to the current workspace") require.Empty(t, syncMessageIDsForRewindTest(t, ctx, db, workspaceID, MessageSyncDirtyScopeCandidate{ ScopeID: scopeID, FrontierSeq: 1, LatestMessageSeq: 2, })) survivorMessage := ScopeMessageRecord{ ID: survivorMessageID, WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 3, WorkspaceGlobalSeq: 30, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "rewind visibility survivor needle", Metadata: []byte(`{}`), CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), } require.NoError(t, repos.Messages.Insert(ctx, survivorMessage)) mustExecMessaging(t, ctx, db, ` UPDATE scopes SET latest_seq = 3, latest_global_seq = 30, latest_message_at = $3, visible_message_count = 2, visible_last_message_at = $3 WHERE id = $1 AND workspace_id = $2 `, scopeID, workspaceID, survivorMessage.CreatedAt) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: survivorMessageID, WorkspaceID: workspaceID, ScopeID: scopeID, GlobalSeq: 30, Content: "rewind visibility survivor needle", CreatedAt: now.Add(3 * time.Second), })) survivorHistory, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Limit: 10, }) require.NoError(t, err) require.Equal(t, []string{survivorMessageID, keptMessageID}, historyMessageIDs(survivorHistory.Messages)) survivorSearch, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 10, CandidateBatchLimit: 10, MaxCandidateBatches: 1, }, 10, func(ctx context.Context, _ MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { return rewindSearchCandidatesFromProjectedDocsForTest(t, ctx, db, ownerMemberID, survivorMessage, keptMessage), nil }) require.NoError(t, err) require.Equal(t, []string{survivorMessageID, keptMessageID}, visibleMessageRefIDs(survivorSearch.Messages)) } func rewindSearchCandidatesFromProjectedDocsForTest(t *testing.T, ctx context.Context, db coredb.Queryer, viewerMemberID string, records ...ScopeMessageRecord) []ScopeMessageVisibilityCandidate { t.Helper() candidates := make([]ScopeMessageVisibilityCandidate, 0, len(records)) for _, record := range records { if countScopeRows(t, ctx, db, "scope_message_search_docs", "message_id = $1", record.ID) == 0 { continue } if countScopeRows(t, ctx, db, "scope_messages", "id = $1", record.ID) == 0 { continue } candidates = append(candidates, scopeSearchCandidate(record, ScopeMembershipRecord{ ScopeID: record.ScopeID, MemberID: viewerMemberID, Visibility: "public", EffectiveFromSeq: 1, })) } return candidates } func ownerQuoteSourceMessageIDsForRewindTest(sources []OwnerQuoteSourceMessage) []string { ids := make([]string, 0, len(sources)) for _, source := range sources { ids = append(ids, source.ID) } return ids } func syncMessageIDsForRewindTest(t *testing.T, ctx context.Context, db coredb.Queryer, workspaceID string, candidate MessageSyncDirtyScopeCandidate) []string { t.Helper() query, args, err := BuildMessageSyncMessagesForScopesQuery(workspaceID, []MessageSyncDirtyScopeCandidate{candidate}, 10) require.NoError(t, err) rows, err := db.Query(ctx, query, args...) require.NoError(t, err) defer rows.Close() ids := []string{} for rows.Next() { var ( id string seq int64 channelID string authorType string senderMemberID string senderUserID string senderAgentID string senderAgentName string userDisplayName string memberDisplayName string userName string userAvatarURL string agentAvatarURL string clientMessageID string messageType string content string metadata []byte createdAt time.Time editedAt sql.NullTime editedByMemberID string editRevision int64 ) require.NoError(t, rows.Scan( &id, &seq, &channelID, &authorType, &senderMemberID, &senderUserID, &senderAgentID, &senderAgentName, &userDisplayName, &memberDisplayName, &userName, &userAvatarURL, &agentAvatarURL, &clientMessageID, &messageType, &content, &metadata, &createdAt, &editedAt, &editedByMemberID, &editRevision, )) ids = append(ids, id) } require.NoError(t, rows.Err()) return ids }