diff --git a/cmd/msgvault/cmd/build_cache.go b/cmd/msgvault/cmd/build_cache.go index d188f77ca..0948f2e08 100644 --- a/cmd/msgvault/cmd/build_cache.go +++ b/cmd/msgvault/cmd/build_cache.go @@ -613,7 +613,8 @@ func derivedDriftOnly(staleness cacheStaleness) bool { staleness.HasConversationTypeDrift || staleness.HasParticipantIdentifierDrift || staleness.HasParticipantDisplayNameDrift) && !staleness.HasNew && !staleness.HasDeleted && - !staleness.HasUpdated && !staleness.HasAccountIdentityDrift + !staleness.HasUpdated && !staleness.HasAccountIdentityDrift && + !staleness.HasDerivedDataDrift } // refreshIdentityDatasetsOnly rebuilds every identity-derived dataset while @@ -691,11 +692,11 @@ func buildCacheLocked( // concurrent identity mutation therefore makes the stamped revision LAG // the store, which HasIdentityDrift detects on the next staleness check — // the cache self-heals. Never move this read after the export. The same - // invariant applies to the account-identity, participant-identifier, and - // participant display-name revisions read alongside it: this full build - // derives all of those datasets fresh from the current store state, so - // stamping a lagging revision here is likewise self-healing — the matching - // staleness check catches it on the next pass. + // invariant applies to the derived-data, account-identity, + // participant-identifier, and participant display-name revisions read + // alongside it: this full build exports those facts from the current store + // state, so stamping a lagging revision here is likewise self-healing — the + // matching staleness check catches it on the next pass. identityStore, err := store.Open(dbPath) if err != nil { return nil, fmt.Errorf("open store for identity export: %w", err) @@ -705,6 +706,11 @@ func buildCacheLocked( _ = identityStore.Close() return nil, fmt.Errorf("read identity revision: %w", err) } + derivedDataRevision, err := identityStore.DerivedDataRevision() + if err != nil { + _ = identityStore.Close() + return nil, fmt.Errorf("read derived-data revision: %w", err) + } accountIdentityRevision, err := identityStore.AccountIdentityRevision() if err != nil { _ = identityStore.Close() @@ -837,6 +843,13 @@ func buildCacheLocked( return nil, fmt.Errorf("inspect attachment MIME schema: %w", err) } sourceSnapshot.hasAttachmentMIME = attachmentMIMEColumnCount > 0 + var attachmentMetadataColumnCount int + if err := sourceSnapshot.QueryRow(` + SELECT COUNT(*) FROM pragma_table_info('attachments') WHERE name = 'attachment_metadata' + `).Scan(&attachmentMetadataColumnCount); err != nil { + return nil, fmt.Errorf("inspect attachment metadata schema: %w", err) + } + sourceSnapshot.hasAttachmentMetadata = attachmentMetadataColumnCount > 0 var messageSourceAttributionColumnCount int if err := sourceSnapshot.QueryRow(` SELECT COUNT(*) FROM pragma_table_info('messages') @@ -981,6 +994,10 @@ func buildCacheLocked( if sourceSnapshot.hasAttachmentMIME { attachmentMIMEExpression = "COALESCE(TRY_CAST(mime_type AS VARCHAR), '') AS mime_type" } + attachmentMetadataExpression := "NULL::VARCHAR AS attachment_metadata" + if sourceSnapshot.hasAttachmentMetadata { + attachmentMetadataExpression = "TRY_CAST(attachment_metadata AS VARCHAR) AS attachment_metadata" + } if err := runExport(tableAttachments, fmt.Sprintf(` COPY ( SELECT @@ -988,13 +1005,15 @@ func buildCacheLocked( message_id, size, COALESCE(TRY_CAST(filename AS VARCHAR), '') as filename, + %s, %s FROM sqlite_db.attachments%s ) TO '%s/%s' ( FORMAT PARQUET, COMPRESSION 'zstd' ) - `, attachmentMIMEExpression, attachmentsFilter, escapedAttachmentsDir, junctionFile)); err != nil { + `, attachmentMIMEExpression, attachmentMetadataExpression, + attachmentsFilter, escapedAttachmentsDir, junctionFile)); err != nil { return nil, fmt.Errorf("export attachments: %w", err) } @@ -1375,6 +1394,7 @@ func buildCacheLocked( LastFailedSyncRunCount: syncCounters.failedRunCount, LastFailedSyncRunIDSum: syncCounters.failedRunIDSum, IdentityRevision: identityRevision, + DerivedDataRevision: derivedDataRevision, AccountIdentityRevision: accountIdentityRevision, ParticipantIdentifierRevision: participantIdentifierRevision, ParticipantDisplayNameRevision: participantDisplayNameRevision, @@ -1599,6 +1619,7 @@ type cacheSourceSnapshot struct { sqliteTx *sql.Tx tmpDir string hasAttachmentMIME bool + hasAttachmentMetadata bool hasMessageSourceAttribution bool hasRecipientEnvelope bool } @@ -1717,10 +1738,16 @@ func (s *cacheSourceSnapshot) PrepareDatasets(names ...string) error { } func (s *cacheSourceSnapshot) tables() []cacheSnapshotTable { - attachmentQuery := "SELECT id, message_id, size, filename, '' AS mime_type FROM attachments" + attachmentMIMEColumn := "'' AS mime_type" if s.hasAttachmentMIME { - attachmentQuery = "SELECT id, message_id, size, filename, mime_type FROM attachments" + attachmentMIMEColumn = "mime_type" + } + attachmentMetadataColumn := "NULL AS attachment_metadata" + if s.hasAttachmentMetadata { + attachmentMetadataColumn = "attachment_metadata" } + attachmentQuery := "SELECT id, message_id, size, filename, " + attachmentMIMEColumn + + ", " + attachmentMetadataColumn + " FROM attachments" recipientEnvelopeColumn := "'' AS email_address" if s.hasRecipientEnvelope { recipientEnvelopeColumn = "email_address" @@ -1748,7 +1775,7 @@ func (s *cacheSourceSnapshot) tables() []cacheSnapshotTable { {"message_labels", "SELECT message_id, label_id FROM message_labels", "types={'message_id': 'BIGINT', 'label_id': 'BIGINT'}"}, {tableAttachments, attachmentQuery, - "types={'id': 'BIGINT', 'message_id': 'BIGINT', 'size': 'BIGINT', 'filename': 'VARCHAR', 'mime_type': 'VARCHAR'}"}, + "types={'id': 'BIGINT', 'message_id': 'BIGINT', 'size': 'BIGINT', 'filename': 'VARCHAR', 'mime_type': 'VARCHAR', 'attachment_metadata': 'VARCHAR'}"}, {tableParticipants, "SELECT id, email_address, domain, display_name, phone_number FROM participants", "types={'id': 'BIGINT', 'email_address': 'VARCHAR', 'domain': 'VARCHAR', 'display_name': 'VARCHAR', 'phone_number': 'VARCHAR'}"}, {"account_identities", "SELECT source_id, address FROM account_identities", diff --git a/cmd/msgvault/cmd/build_cache_test.go b/cmd/msgvault/cmd/build_cache_test.go index a79aabd94..dfc65b47a 100644 --- a/cmd/msgvault/cmd/build_cache_test.go +++ b/cmd/msgvault/cmd/build_cache_test.go @@ -2111,6 +2111,63 @@ func TestBuildCache_UTF8Handling(t *testing.T) { assert.Equal("Test émoji 🎉 and unicode", subject, "unicode should be preserved") } +func TestBuildCacheExportsAttachmentMetadataForRawQuery(t *testing.T) { + for _, tc := range []struct { + name string + forceCSV bool + }{ + {name: "sqlite scanner"}, + {name: "CSV fallback", forceCSV: true}, + } { + t.Run(tc.name, func(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + if tc.forceCSV { + t.Setenv("MSGVAULT_FORCE_CSV_SNAPSHOT", "1") + } else { + t.Setenv("MSGVAULT_FORCE_CSV_SNAPSHOT", "") + } + + tmpDir := setupTestSQLite(t) + dbPath := filepath.Join(tmpDir, "test.db") + analyticsDir := filepath.Join(tmpDir, "analytics") + db, err := sql.Open("sqlite3", dbPath) + require.NoError(err, "open SQLite fixture") + _, err = db.Exec(`ALTER TABLE attachments ADD COLUMN attachment_metadata JSON`) + require.NoError(err, "add attachment metadata column") + _, err = db.Exec(`UPDATE attachments SET attachment_metadata = '{"shared_url":"https://example.com/post"}' WHERE id = 1`) + require.NoError(err, "set link-preview metadata") + _, err = db.Exec(`UPDATE messages SET message_type = 'beeper' WHERE id = 2`) + require.NoError(err, "mark fixture message as Beeper") + require.NoError(db.Close(), "close SQLite fixture") + + _, err = buildCache(dbPath, analyticsDir, true) + require.NoError(err, "build analytics cache") + engine, err := query.NewDuckDBEngine(analyticsDir, "", nil) + require.NoError(err, "open analytics query engine") + defer func() { _ = engine.Close() }() + + result, err := engine.QuerySQL(context.Background(), ` + SELECT COALESCE(a.attachment_metadata IS NOT NULL, 0) AS is_share, + COUNT(*), SUM(a.size) + FROM attachments a + JOIN messages m ON m.id = a.message_id + WHERE m.message_type = 'beeper' + GROUP BY is_share + ORDER BY is_share`) + require.NoError(err, "run documented link-preview query") + assert.Equal([]string{"is_share", "count_star()", "sum(a.size)"}, result.Columns) + require.Len(result.Rows, 2) + assert.Equal("0", fmt.Sprint(result.Rows[0][0])) + assert.Equal("1", fmt.Sprint(result.Rows[0][1])) + assert.Equal("5000", fmt.Sprint(result.Rows[0][2])) + assert.Equal("1", fmt.Sprint(result.Rows[1][0])) + assert.Equal("1", fmt.Sprint(result.Rows[1][1])) + assert.Equal("10000", fmt.Sprint(result.Rows[1][2])) + }) + } +} + // TestBuildCache_EmptyDatabase tests handling of empty database. func TestBuildCache_EmptyDatabase(t *testing.T) { require := require.New(t) @@ -3302,8 +3359,7 @@ func TestCacheNeedsBuild_IgnoresAlreadyProcessedUpdatedSyncRun(t *testing.T) { // schema version other than the current one now forces a full rebuild. func TestCacheNeedsBuild_SchemaVersionMismatch(t *testing.T) { require := require.New(t) - require.Equal(18, cacheSchemaVersion, - "participant directory revisions require a one-time cache rebuild at v18") + require.Equal(19, cacheSchemaVersion, "attachment metadata requires cache v19") tmpDir := setupTestSQLiteEmpty(t) dbPath := filepath.Join(tmpDir, "test.db") @@ -3338,7 +3394,7 @@ func TestCacheNeedsBuild_SchemaVersionMismatch(t *testing.T) { require.False(result.Skipped, "schema mismatch must execute a full rebuild") upgraded, err := query.ReadCacheSyncState(analyticsDir) require.NoError(err, "read upgraded cache state") - require.Equal(18, upgraded.SchemaVersion) + require.Equal(19, upgraded.SchemaVersion) require.NoFileExists(filepath.Join(analyticsDir, tableParticipantIdentifiers, "data.parquet"), "full rebuild must replace rather than extend the v11 identifier dataset") identifierParquet := filepath.Join(analyticsDir, tableParticipantIdentifiers, "participant_identifiers.parquet") diff --git a/cmd/msgvault/cmd/cache_derived.go b/cmd/msgvault/cmd/cache_derived.go index 5538f0dc9..6f7a1e5e5 100644 --- a/cmd/msgvault/cmd/cache_derived.go +++ b/cmd/msgvault/cmd/cache_derived.go @@ -69,6 +69,16 @@ func refreshDerivedDatasetsOnly( _ = st.Close() return nil, fmt.Errorf("read identity revision: %w", err) } + derivedDataRevision, err := st.DerivedDataRevision() + if err != nil { + _ = st.Close() + return nil, fmt.Errorf("read derived-data revision: %w", err) + } + if derivedDataRevision != state.DerivedDataRevision { + _ = st.Close() + return nil, fmt.Errorf("%w: derived-data revision changed", + ErrDerivedRefreshRequiresFullBuild) + } accountIdentityRevision, err := st.AccountIdentityRevision() if err != nil { _ = st.Close() diff --git a/cmd/msgvault/cmd/cache_refresh_test.go b/cmd/msgvault/cmd/cache_refresh_test.go index 401f90b90..98cae8f7d 100644 --- a/cmd/msgvault/cmd/cache_refresh_test.go +++ b/cmd/msgvault/cmd/cache_refresh_test.go @@ -2,8 +2,10 @@ package cmd import ( "context" + "database/sql" "encoding/json" "errors" + "fmt" "io/fs" "os" "path/filepath" @@ -19,6 +21,7 @@ import ( "go.kenn.io/msgvault/internal/identityindex" "go.kenn.io/msgvault/internal/oauth" "go.kenn.io/msgvault/internal/query" + "go.kenn.io/msgvault/internal/rederive" "go.kenn.io/msgvault/internal/store" ) @@ -332,6 +335,119 @@ func TestRebuildCacheAfterWriteReturnsError(t *testing.T) { require.ErrorContains(err, "refresh analytics cache") } +func TestRebuildCacheAfterDerivedRepairRefreshesCurrentCache(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + tmpDir := t.TempDir() + dbPath := filepath.Join(tmpDir, "msgvault.db") + + savedCfg := cfg + t.Cleanup(func() { cfg = savedCfg }) + cfg = &config.Config{HomeDir: tmpDir, Data: config.DataConfig{DataDir: tmpDir}} + analyticsDir := cfg.AnalyticsDir() + + st, err := store.Open(dbPath) + require.NoError(err) + require.NoError(st.InitSchema()) + source, err := st.GetOrCreateSource("beeper", "signal") + require.NoError(err) + conversationID, err := st.EnsureConversationWithType( + source.ID, "!cache-repair:example.org", "direct_chat", "Cache repair", + ) + require.NoError(err) + sentAt := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + messageID, err := st.UpsertMessage(&store.Message{ + ConversationID: conversationID, + SourceID: source.ID, + SourceMessageID: "repair-cache-1", + MessageType: "beeper", + SentAt: sql.NullTime{Time: sentAt, Valid: true}, + ReceivedAt: sql.NullTime{Time: sentAt, Valid: true}, + Snippet: sql.NullString{String: "stale snippet", Valid: true}, + HasAttachments: true, + AttachmentCount: 1, + }) + require.NoError(err) + require.NoError(st.UpsertMessageBody( + messageID, + sql.NullString{String: "stale body", Valid: true}, + sql.NullString{}, + )) + raw, err := json.Marshal(map[string]any{ + "id": "repair-cache-1", + "chatID": "!cache-repair:example.org", + "accountID": "signal", + "senderID": "@user-a:example.org", + "senderName": "User A", + "timestamp": sentAt, + "type": "IMAGE", + "text": "https://example.com/post", + "attachments": []map[string]any{{ + "id": "mxc://example.org/share", "type": "img", "mimeType": "image/jpeg", + }}, + }) + require.NoError(err) + require.NoError(st.UpsertMessageRawWithFormat(messageID, raw, "beeper_json")) + require.NoError(st.ReplaceMessageBeeperAttachments(messageID, []store.AttachmentRef{{ + MimeType: "image/jpeg", + StoragePath: "mxc://example.org/share", + SourceAttachmentID: "beeper:mxc://example.org/share", + MediaType: "image", + }})) + require.NoError(st.Close()) + + _, err = buildCache(dbPath, analyticsDir, true) + require.NoError(err, "build current-schema cache before repair") + initialState, err := query.ReadCacheSyncState(analyticsDir) + require.NoError(err) + assert.Equal(query.CacheSchemaVersion, initialState.SchemaVersion) + assert.Zero(initialState.DerivedDataRevision) + + readCached := func() (string, any) { + t.Helper() + engine, openErr := query.NewDuckDBEngine(analyticsDir, "", nil) + require.NoError(openErr) + result, queryErr := engine.QuerySQL(context.Background(), ` + SELECT m.snippet, a.attachment_metadata + FROM messages m + JOIN attachments a ON a.message_id = m.id + WHERE m.source_message_id = 'repair-cache-1'`) + closeErr := engine.Close() + require.NoError(queryErr) + require.NoError(closeErr) + require.Len(result.Rows, 1) + return fmt.Sprint(result.Rows[0][0]), result.Rows[0][1] + } + + beforeSnippet, beforeMetadata := readCached() + assert.Equal("stale snippet", beforeSnippet) + assert.Nil(beforeMetadata) + + st, err = store.Open(dbPath) + require.NoError(err) + sum, err := rederive.Run( + context.Background(), st, "beeper", source.Identifier, source.ID, nil, + ) + require.NoError(err) + require.Zero(sum.Errors) + require.NoError(st.Close()) + + staleness := cacheNeedsBuild(dbPath, analyticsDir) + require.True(staleness.NeedsBuild) + assert.True(staleness.HasDerivedDataDrift) + assert.True(staleness.FullRebuild, + "an incremental append cannot replace already-cached repaired rows") + + require.NoError(rebuildCacheAfterWrite(dbPath)) + repairedState, err := query.ReadCacheSyncState(analyticsDir) + require.NoError(err) + assert.Equal(int64(1), repairedState.DerivedDataRevision) + afterSnippet, afterMetadata := readCached() + assert.Equal("https://example.com/post", afterSnippet) + require.NotNil(afterMetadata) + assert.JSONEq(`{"shared_url":"https://example.com/post"}`, fmt.Sprint(afterMetadata)) +} + func TestScheduledCacheRefreshSkipsWhenAutoBuildCacheDisabled(t *testing.T) { require := require.New(t) assert := assert.New(t) diff --git a/cmd/msgvault/cmd/cache_staleness.go b/cmd/msgvault/cmd/cache_staleness.go index 6b20e7e09..44272c54a 100644 --- a/cmd/msgvault/cmd/cache_staleness.go +++ b/cmd/msgvault/cmd/cache_staleness.go @@ -23,6 +23,11 @@ type cacheStaleness struct { // index-only refresh applies must check HasAccountIdentityDrift too — // see derivedDriftOnly in build_cache.go. HasIdentityDrift bool + // HasDerivedDataDrift signals an offline repair rewrote existing message + // or attachment facts already inside the committed cache watermark. These + // facts require a full rebuild; neither incremental append nor the + // identity-only refresh can replace them. + HasDerivedDataDrift bool // HasConversationParticipantDrift signals conversation membership changed // for a conversation already represented by the committed message // watermark. The index-only refresh can rebuild relationship_activity and @@ -294,6 +299,19 @@ func cacheNeedsBuildLocked(dbPath, analyticsDir string) cacheStaleness { } } + derivedDataRevision, err := db.DerivedDataRevision() + if err != nil { + return cacheStaleness{ + NeedsBuild: true, FullRebuild: true, + Reason: "cannot verify derived-data revision", + } + } + if derivedDataRevision != state.DerivedDataRevision { + result.HasDerivedDataDrift = true + result.FullRebuild = true + reasons = append(reasons, "derived message data changed") + } + // Account-identity drift covers identity mutations that invalidate baked // message data: confirming or removing a confirmed "me" address via // AddAccountIdentity/RemoveAccountIdentity, and participant merges via diff --git a/cmd/msgvault/cmd/repair_derived.go b/cmd/msgvault/cmd/repair_derived.go new file mode 100644 index 000000000..3a397bac1 --- /dev/null +++ b/cmd/msgvault/cmd/repair_derived.go @@ -0,0 +1,142 @@ +package cmd + +import ( + "errors" + "fmt" + "strings" + "time" + + "github.com/spf13/cobra" + "go.kenn.io/msgvault/internal/rederive" + "go.kenn.io/msgvault/internal/store" +) + +var ( + repairDerivedSourceTypes []string + repairDerivedIdentifiers []string +) + +func newRepairDerivedCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "repair-derived", + Short: "Re-derive stored message text and metadata from archived payloads", + Long: `Re-derive stored message columns from the payloads archived with them. + +Message bodies, snippets, the search index, and attachment metadata are computed +from a provider's payload when a message is imported, so improving how they are +derived leaves already-archived rows stale. This command recomputes them from +the verbatim payload stored alongside every message. + +Syncing already heals an archive on its own — each source re-derives once, on its +next sync — so this is for repairing on demand instead of waiting, or for +re-running after an interrupted pass. It works entirely against the local +archive: no provider connection is needed, and messages the provider no longer +holds are repaired too. Only derived columns are rewritten; raw payloads, +downloaded media, and sync cursors are untouched, so it is idempotent. + +Examples: + msgvault repair-derived + msgvault repair-derived --source-type beeper + msgvault repair-derived --source-type beeper --identifier instagramgo`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + if !isDaemonCLISubprocess() { + return runDaemonCLICommandHTTPFromCobra(cmd, args) + } + + s, cleanup, err := openWritableStoreAndInitForIngest() + if err != nil { + return err + } + defer cleanup() + ctx, stop := withInterruptCancel(cmd, "\nInterrupted. Stopping...") + defer stop() + + sources, err := repairDerivedTargets(s) + if err != nil { + return err + } + if len(sources) == 0 { + _, _ = fmt.Fprintf(cmd.OutOrStdout(), + "No matching sources with a re-derivation pass (available: %s)\n", + strings.Join(rederive.SourceTypes(), ", ")) + return nil + } + + for _, src := range sources { + label := src.SourceType + "/" + src.Identifier + progress := func(msg string) { _, _ = fmt.Fprintf(cmd.OutOrStdout(), " %s: %s\n", label, msg) } + sum, rerr := rederive.Run(ctx, s, src.SourceType, src.Identifier, src.ID, progress) + if ctx.Err() != nil { + _, _ = fmt.Fprintln(cmd.OutOrStdout(), "\nInterrupted — re-run repair-derived to finish (idempotent).") + return rebuildCacheAfterWrite(cfg.DatabaseDSN()) + } + if rerr != nil { + return errors.Join( + fmt.Errorf("repair failed for %s: %w", label, rerr), + rebuildCacheAfterWrite(cfg.DatabaseDSN()), + ) + } + _, _ = fmt.Fprintf(cmd.OutOrStdout(), + "%s: %d messages scanned, %d bodies rewritten, %d attachments tagged (%s)\n", + label, sum.MessagesScanned, sum.BodiesRewritten, sum.AttachmentsTagged, + sum.Duration.Round(time.Second)) + if sum.Undecodable > 0 { + _, _ = fmt.Fprintf(cmd.OutOrStdout(), " %d archived payloads could not be decoded — left unchanged\n", sum.Undecodable) + } + if sum.Errors > 0 { + _, _ = fmt.Fprintf(cmd.OutOrStdout(), " %d errors — re-run to retry\n", sum.Errors) + } + } + + return rebuildCacheAfterWrite(cfg.DatabaseDSN()) + }, + } + cmd.Flags().StringArrayVar(&repairDerivedSourceTypes, "source-type", nil, + "source type to repair (repeatable; default: every type with a re-derivation pass)") + cmd.Flags().StringArrayVar(&repairDerivedIdentifiers, "identifier", nil, + "source identifier to repair (repeatable; default: all matching sources)") + return cmd +} + +// repairDerivedTargets resolves the sources this run should repair: those whose +// type has a registered pass, narrowed by the --source-type and --identifier +// flags. An unknown source type is an error rather than a silent no-op, so a +// typo does not look like a clean run. +func repairDerivedTargets(s *store.Store) ([]*store.Source, error) { + wantType := map[string]bool{} + for _, t := range repairDerivedSourceTypes { + if _, _, ok := rederive.Lookup(t); !ok { + return nil, fmt.Errorf("no re-derivation pass for source type %q (available: %s)", + t, strings.Join(rederive.SourceTypes(), ", ")) + } + wantType[t] = true + } + wantID := map[string]bool{} + for _, id := range repairDerivedIdentifiers { + wantID[id] = true + } + + all, err := s.ListSources("") + if err != nil { + return nil, err + } + var out []*store.Source + for _, src := range all { + if _, _, ok := rederive.Lookup(src.SourceType); !ok { + continue + } + if len(wantType) > 0 && !wantType[src.SourceType] { + continue + } + if len(wantID) > 0 && !wantID[src.Identifier] { + continue + } + out = append(out, src) + } + return out, nil +} + +func init() { + rootCmd.AddCommand(newRepairDerivedCmd()) +} diff --git a/cmd/msgvault/cmd/show_message.go b/cmd/msgvault/cmd/show_message.go index 39f3fbd7e..2648a852d 100644 --- a/cmd/msgvault/cmd/show_message.go +++ b/cmd/msgvault/cmd/show_message.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" "go.kenn.io/msgvault/internal/query" "go.kenn.io/msgvault/internal/store" + "go.kenn.io/msgvault/internal/textutil" ) var ( @@ -144,9 +145,9 @@ func outputMessageText(msg *query.MessageDetail) error { // Body fmt.Println("\n═══════════════════════════════════════════════════════════════════════════════") if msg.BodyText != "" { - fmt.Println(msg.BodyText) + fmt.Println(textutil.SanitizeTerminalMultiline(msg.BodyText)) } else if msg.Snippet != "" { - fmt.Printf("[No body text available. Snippet: %s]\n", msg.Snippet) + fmt.Printf("[No body text available. Snippet: %s]\n", textutil.SanitizeTerminal(msg.Snippet)) } else { fmt.Println("[No body content available]") } diff --git a/cmd/msgvault/cmd/show_message_test.go b/cmd/msgvault/cmd/show_message_test.go index e66abfec0..f54d702f3 100644 --- a/cmd/msgvault/cmd/show_message_test.go +++ b/cmd/msgvault/cmd/show_message_test.go @@ -5,14 +5,34 @@ import ( "net/http/httptest" "sync/atomic" "testing" + "time" "github.com/spf13/cobra" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.kenn.io/kit/daemon" "go.kenn.io/msgvault/internal/config" + "go.kenn.io/msgvault/internal/query" ) +func TestOutputMessageTextSanitizesMultilineBody(t *testing.T) { + assert := assert.New(t) + done := captureStdout(t) + err := outputMessageText(&query.MessageDetail{ + ID: 42, + SourceMessageID: "remote-42", + SentAt: time.Date(2024, time.January, 2, 3, 4, 5, 0, time.UTC), + BodyText: "first line\n\x1b]52;c;evil\x07second line\u009b", + }) + out := done() + + require.NoError(t, err) + assert.Contains(out, "first line\nsecond line") + assert.NotContains(out, "\x1b") + assert.NotContains(out, "\x07") + assert.NotContains(out, "\u009b") +} + func TestShowMessageUsesLocalDaemonHTTPAndPreservesTextOutput(t *testing.T) { require := require.New(t) assert := assert.New(t) diff --git a/cmd/msgvault/cmd/sync_beeper.go b/cmd/msgvault/cmd/sync_beeper.go index 9852e8860..dbe890e26 100644 --- a/cmd/msgvault/cmd/sync_beeper.go +++ b/cmd/msgvault/cmd/sync_beeper.go @@ -114,6 +114,9 @@ func printBeeperSummary(cmd *cobra.Command, accountID string, sum *beeper.Import if sum.ReactionsRefreshed > 0 { _, _ = fmt.Fprintf(cmd.OutOrStdout(), ", %d reactions refreshed", sum.ReactionsRefreshed) } + if sum.ChatsReopened > 0 { + _, _ = fmt.Fprintf(cmd.OutOrStdout(), ", %d chats reopened for backfilled history", sum.ChatsReopened) + } if sum.AttachmentsDownloaded > 0 { _, _ = fmt.Fprintf(cmd.OutOrStdout(), ", %d attachments", sum.AttachmentsDownloaded) } diff --git a/docs/usage/beeper.md b/docs/usage/beeper.md index 546b45aaa..f00079899 100644 --- a/docs/usage/beeper.md +++ b/docs/usage/beeper.md @@ -100,8 +100,51 @@ are only picked up by `--full` runs. so nothing is lost even where msgvault's relational model is narrower. Because Beeper's API serves what Beeper Desktop has synced locally, archive -depth equals your local Beeper history: a freshly added Beeper account may -only have recent messages until Beeper finishes its own backfill. +depth equals your local Beeper history: a freshly added Beeper account may only +have recent messages until Beeper finishes its own backfill. That backfill can +land hours or weeks later, behind history msgvault has already walked, so once a +day each sync re-checks the oldest end of every completed chat and resumes the +backfill wherever Beeper has since filled more in. Nothing is needed to trigger +this, and the run reports how many chats it reopened. + +## Repairing derived data + +Message bodies, snippets, the search index, and attachment classification are +derived from the API payload at import time, so improvements to how they are +derived do not reach messages already archived. + +Nothing is needed to pick those up: each account re-derives itself once, on its +next sync, from the verbatim JSON stored with every message. The sync reports +how many rows it repaired. To repair on demand instead of waiting — or to finish +an interrupted pass — run it directly: + +```bash +msgvault repair-derived --source-type beeper +msgvault repair-derived --source-type beeper --identifier instagramgo +``` + +It needs no Beeper Desktop connection, repairs messages Beeper no longer holds, +and rewrites only derived columns — raw payloads, downloaded media, and sync +cursors are untouched, so it is idempotent. + +## Link previews + +Media that arrives as a forwarded link preview — an Instagram reel, an x.com +post — is recorded with the URL it previews in `attachments.attachment_metadata` +(`{"shared_url": "..."}`), while media a sender composed has none. This tells a +photo a friend took apart from a public post they forwarded, which matters +because forwarded previews can dominate an Instagram archive's bytes while +remaining recoverable from the URL. Downloads are unaffected: everything is +still archived. To see the split: + +```sql +SELECT COALESCE(a.attachment_metadata IS NOT NULL, 0) AS is_share, + COUNT(*), SUM(a.size) +FROM attachments a +JOIN messages m ON m.id = a.message_id +WHERE m.message_type = 'beeper' +GROUP BY is_share; +``` ## Scheduled sync diff --git a/internal/api/cli_handlers.go b/internal/api/cli_handlers.go index bd83be946..18ab66eda 100644 --- a/internal/api/cli_handlers.go +++ b/internal/api/cli_handlers.go @@ -1402,6 +1402,7 @@ func cliRunCommandAllowed(args []string) bool { "repair-dates", "repack-attachments", "remove-account", + "repair-derived", "show-deletion", "sync-beeper", "sync-calendar", diff --git a/internal/beeper/anchors.go b/internal/beeper/anchors.go index b22ab72c2..f6a854de2 100644 --- a/internal/beeper/anchors.go +++ b/internal/beeper/anchors.go @@ -97,7 +97,7 @@ func (imp *Importer) verifyArchivedSample(ctx context.Context, sourceID int64) e // from prior runs when this run enumerated none (quiet account) — the guard // must not stay weakened just because nothing happened lately. Best-effort; a // failure leaves the remaining slots for the next run to fill. -func (imp *Importer) rearmAnchors(ctx context.Context, chats []Chat, state *SyncState) { +func (imp *Importer) rearmAnchors(ctx context.Context, chats []chatVisit, state *SyncState) { if len(state.Anchors) >= maxAnchors { return } @@ -157,7 +157,7 @@ func anchorFrom(chatID string, items []Message) *AnchorProbe { var best *Message for i := range items { m := &items[i] - if m.Type == "REACTION" || m.IsDeleted || m.IsHidden { + if !persistsMessageRow(m) { continue } if best == nil || m.Timestamp.After(best.Timestamp) { diff --git a/internal/beeper/fake_server_test.go b/internal/beeper/fake_server_test.go index 60057b4e4..fef4900d8 100644 --- a/internal/beeper/fake_server_test.go +++ b/internal/beeper/fake_server_test.go @@ -94,7 +94,10 @@ type fakeBeeper struct { cancelAfterPages int cancelFn func() pagesServed int - reqs []string // "PATH?QUERY" per request, in order + // cancelOnMessageListChatID invokes cancelFn once after serving a + // message-list response for the named chat. + cancelOnMessageListChatID string + reqs []string // "PATH?QUERY" per request, in order } func newFakeBeeper(t *testing.T) *fakeBeeper { @@ -185,6 +188,22 @@ func (f *fakeBeeper) appendMsg(chatID string, m fakeMsg) { require.Failf(f.t, "appendMsg failed", "unknown chat %s", chatID) } +// prependMsgs adds messages behind a chat's existing oldest message, as Beeper +// Desktop does when it finishes backfilling a network's older history. The +// chat's lastActivity deliberately does not move: that is what makes the new +// history invisible to activity-filtered enumeration and head reconciliation. +func (f *fakeBeeper) prependMsgs(chatID string, msgs ...fakeMsg) { + f.mu.Lock() + defer f.mu.Unlock() + for _, ch := range f.chats { + if ch.ID == chatID { + ch.Msgs = append(append([]fakeMsg{}, msgs...), ch.Msgs...) + return + } + } + require.Failf(f.t, "prependMsgs failed", "unknown chat %s", chatID) +} + // requests returns the request log ("PATH?QUERY" entries). func (f *fakeBeeper) requests() []string { f.mu.Lock() @@ -198,6 +217,13 @@ func (f *fakeBeeper) resetRequests() { f.reqs = nil } +func (f *fakeBeeper) cancelMessageListFor(chatID string, cancelFn func()) { + f.mu.Lock() + defer f.mu.Unlock() + f.cancelOnMessageListChatID = chatID + f.cancelFn = cancelFn +} + func (f *fakeBeeper) server() *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f.mu.Lock() @@ -366,6 +392,10 @@ func (f *fakeBeeper) writeMessages(w http.ResponseWriter, r *http.Request, chatI if f.cancelAfterPages > 0 && f.pagesServed == f.cancelAfterPages && f.cancelFn != nil { defer f.cancelFn() // after the response is written } + if f.cancelOnMessageListChatID == chatID && f.cancelFn != nil { + defer f.cancelFn() // after the response is written + f.cancelOnMessageListChatID = "" + } var ch *fakeChat for _, c := range f.chats { if c.ID == chatID { diff --git a/internal/beeper/importer.go b/internal/beeper/importer.go index fb1e2f5b8..e97c711ef 100644 --- a/internal/beeper/importer.go +++ b/internal/beeper/importer.go @@ -8,6 +8,7 @@ import ( "fmt" "time" + "go.kenn.io/msgvault/internal/rederive" "go.kenn.io/msgvault/internal/store" ) @@ -33,8 +34,20 @@ const ( reconcileWindow = 24 * time.Hour // maxReconcilePages caps the reconciliation walk for pathologically busy chats. maxReconcilePages = 50 + // maxTailProbePages bounds scans through runs of non-content events. Hitting + // the bound conservatively reopens the chat so backfill can keep progressing. + maxTailProbePages = 20 ) +// tailScanInterval throttles the completed-chat tail probe (see tailScanDue). +// Beeper Desktop backfills a network's older history over hours-to-weeks after +// it is linked; those messages arrive with old timestamps, so they neither +// advance a chat's lastActivity nor fall in the reconcile window, and a chat +// already marked done would never see them. Probing costs one request per +// completed chat, so it runs at most daily. +// A variable so tests can disable the throttle. +var tailScanInterval = 24 * time.Hour + // chatScope carries per-chat state through the persist call chain: the chat // and store IDs, the run options, and the chat's cursor state (whose // PendingReplies buffer persistMessage appends to, keeping checkpoints @@ -48,6 +61,18 @@ type chatScope struct { cs *ChatState membershipComplete bool budgetUsed int + // tailScan asks a completed chat to re-probe the oldest end of its history + // before settling into the incremental path (see tailScanInterval). + tailScan bool +} + +// chatVisit records why a chat was enumerated. tailOnly means the chat would +// have been excluded by the normal activity filter and is present solely for +// the completed-history probe. +type chatVisit struct { + Chat + + tailOnly bool } func (cc *chatScope) limitReached() bool { @@ -117,6 +142,19 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum state.Anchors = anchors } + // Heal rows derived by an older build before syncing new ones, so an + // upgraded archive converges without the user knowing to run a repair. + // Ledger-gated, so this costs one indexed lookup on every later run. + rsum, ran, rerr := rederive.RunIfStale(ctx, imp.store, sourceTypeBeeper, opts.AccountID, src.ID, opts.Progress) + if rerr != nil { + return nil, rerr + } + if ran && rsum != nil { + sum.BodiesRepaired = rsum.BodiesRewritten + sum.AttachmentsRetagged = rsum.AttachmentsTagged + sum.Errors += rsum.Errors + } + syncID, err := imp.store.StartSync(src.ID, sourceTypeBeeper) if err != nil { return nil, err @@ -141,8 +179,13 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum return sum, err } + // A tail scan must see every chat, not just recently-active ones: a chat + // gains backfilled history without its lastActivity moving, so the usual + // enumeration filter would skip exactly the chats worth probing. + tailScan := opts.Full || tailScanDue(state.LastTailScan, start) + reconcileCutoff := start.Add(-reconcileWindow) - chats, err := imp.enumerateChats(ctx, syncID, opts, state, reconcileCutoff, sum) + chats, err := imp.enumerateChats(ctx, syncID, opts, state, reconcileCutoff, tailScan, sum) if err != nil { return sum, err } @@ -153,7 +196,8 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum maxActivity := parseWatermark(state.ListWatermark) total := len(chats) for idx := range chats { - ch := &chats[idx] + visit := &chats[idx] + ch := &visit.Chat if err = ctx.Err(); err != nil { return sum, err } @@ -161,7 +205,7 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum maxActivity = ch.LastActivity } var convCount int64 - convCount, err = imp.syncChat(ctx, syncID, src.ID, ch, opts, state, reconcileCutoff, sum) + convCount, err = imp.syncChat(ctx, syncID, src.ID, ch, opts, state, reconcileCutoff, tailScan, visit.tailOnly, sum) if err != nil { return sum, err } @@ -172,12 +216,21 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum // Flush checkpoint so an interrupted run can resume from this point. imp.checkpoint(syncID, state, sum) } + if err = ctx.Err(); err != nil { + return sum, err + } // Advance the discovery watermark only for fetch-clean runs: a fetch error // means some chat's messages are still missing, so it must stay // discoverable by the next run's lastActivityAfter filter. if sum.FetchErrors == 0 && !maxActivity.IsZero() { state.ListWatermark = formatWatermark(maxActivity) } + // Record the scan only on a fetch-clean run: a run that failed partway + // through may not have probed every chat, and re-probing costs one request + // per chat rather than any lost data. + if tailScan && sum.FetchErrors == 0 { + state.LastTailScan = formatWatermark(start) + } // Never complete a run under-anchored: incremental-only runs skip the // backfill path that normally arms probes, and persisting none would @@ -210,25 +263,22 @@ func (imp *Importer) Import(ctx context.Context, opts ImportOptions) (*ImportSum // enumerateChats lists the chats this run must visit: every chat active in // the discovery overlap or reconciliation window (all chats on first/full // runs), plus any chat whose backfill is unfinished even without new activity. -func (imp *Importer) enumerateChats(ctx context.Context, syncID int64, opts ImportOptions, state *SyncState, reconcileCutoff time.Time, sum *ImportSummary) ([]Chat, error) { +func (imp *Importer) enumerateChats(ctx context.Context, syncID int64, opts ImportOptions, state *SyncState, reconcileCutoff time.Time, tailScan bool, sum *ImportSummary) ([]chatVisit, error) { params := SearchChatsParams{AccountID: opts.AccountID} - if !opts.Full && state.ListWatermark != "" { - if wm := parseWatermark(state.ListWatermark); !wm.IsZero() { - // Overlap by an hour so clock skew or a mid-listing crash cannot - // permanently hide a chat from enumeration. Also include every chat - // inside the reconciliation window so in-place changes are revisited - // even when their LastActivity did not advance. - params.LastActivityAfter = wm.Add(-time.Hour) - if reconcileCutoff.Before(params.LastActivityAfter) { - params.LastActivityAfter = reconcileCutoff - } - } + activityCutoff := chatActivityCutoff(opts, state, reconcileCutoff) + if !tailScan { + params.LastActivityAfter = activityCutoff } - var chats []Chat + var chats []chatVisit seen := map[string]bool{} err := imp.client.AllChats(ctx, params, func(ch Chat) error { seen[ch.ID] = true - chats = append(chats, ch) + tailOnly := tailScan && !activityCutoff.IsZero() && !ch.LastActivity.After(activityCutoff) + if cs := state.Chats[ch.ID]; cs != nil && !cs.Done { + // Unfinished backfills are included independently of activity. + tailOnly = false + } + chats = append(chats, chatVisit{Chat: ch, tailOnly: tailOnly}) return nil }) if err != nil { @@ -256,15 +306,36 @@ func (imp *Importer) enumerateChats(ctx context.Context, syncID int64, opts Impo sum.Errors++ continue } - chats = append(chats, *detail) + chats = append(chats, chatVisit{Chat: *detail}) } return chats, nil } +// chatActivityCutoff returns the filter a normal incremental run would send +// to chat discovery. Tail scans omit the API filter but retain this value to +// distinguish active work from chats enumerated solely for probing. +func chatActivityCutoff(opts ImportOptions, state *SyncState, reconcileCutoff time.Time) time.Time { + if opts.Full { + return time.Time{} + } + wm := parseWatermark(state.ListWatermark) + if wm.IsZero() { + return time.Time{} + } + // Overlap by an hour so clock skew or a mid-listing crash cannot hide a + // chat, and include the reconciliation window for in-place changes whose + // LastActivity did not advance. + cutoff := wm.Add(-time.Hour) + if reconcileCutoff.Before(cutoff) { + cutoff = reconcileCutoff + } + return cutoff +} + // syncChat ensures the conversation and its participants, then backfills or // incrementally extends the chat's messages. Returns the number of messages // processed for this chat. -func (imp *Importer) syncChat(ctx context.Context, syncID, sourceID int64, ch *Chat, opts ImportOptions, state *SyncState, reconcileCutoff time.Time, sum *ImportSummary) (int64, error) { +func (imp *Importer) syncChat(ctx context.Context, syncID, sourceID int64, ch *Chat, opts ImportOptions, state *SyncState, reconcileCutoff time.Time, tailScan, tailOnly bool, sum *ImportSummary) (int64, error) { convID, membershipComplete, err := imp.ensureConversation(ctx, syncID, sourceID, ch, sum) if err != nil { return 0, err @@ -274,9 +345,27 @@ func (imp *Importer) syncChat(ctx context.Context, syncID, sourceID int64, ch *C cc := &chatScope{ chatID: ch.ID, convID: convID, sourceID: sourceID, syncID: syncID, opts: opts, cs: cs, membershipComplete: membershipComplete, + tailScan: tailScan, } before := sum.MessagesProcessed + // Re-open a completed chat whose oldest end has grown since it was walked; + // clearing Done routes it back through the backfill path below. + if cs.Done && tailScan { + var reopened bool + reopened, err = imp.probeChatTail(ctx, cc, sum) + if err != nil { + return sum.MessagesProcessed - before, err + } + if tailOnly && !reopened && cs.Newest != "" { + // This quiet chat was enumerated only for the probe. With no new + // history, its incremental and reconciliation paths have no work. + // Cursorless chats still need the empty-chat recovery below. + imp.flushReplies(cc, sum) + return sum.MessagesProcessed - before, nil + } + } + // A chat that was empty when backfilled has Done set but no incremental // cursor; re-walk it from scratch (cheap) so its first messages are seen. if !cs.Done || cs.Newest == "" { @@ -354,6 +443,87 @@ func (imp *Importer) ensureConversation(ctx context.Context, syncID, sourceID in return convID, membershipComplete, nil } +// probeChatTail searches past a completed chat's oldest cursor and clears Done +// when it finds messages the archive has never seen, so the backfill resumes +// into history Beeper added after the chat was first walked. +// +// The archive is consulted rather than just trusting a non-empty page: near the +// beginning of history the API re-serves the tail it already returned (the same +// misbehaviour backfillChat's recentIDWindow defends against), so a page of +// familiar messages must leave the chat done or every scan would re-walk it. +// +// Best-effort except for context cancellation, which must abort the run so the +// scan remains due. Other probe failures leave the chat completed and are not +// counted as fetch errors: the messages they would find are ones the archive +// has never had, so deferring them to the next scan loses nothing captured. +func (imp *Importer) probeChatTail(ctx context.Context, cc *chatScope, sum *ImportSummary) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err + } + cursor := cc.cs.Oldest + if cursor == "" { + return false, nil + } + recent := newRecentIDWindow(recentIDWindowPages) + for range maxTailProbePages { + page, err := imp.client.ListMessagesPage(ctx, cc.chatID, cursor, "before") + if ctxErr := ctx.Err(); ctxErr != nil { + return false, ctxErr + } + if err != nil || len(page.Items) == 0 { + return false, nil //nolint:nilerr // non-cancellation probe failures are deferred to the next scan + } + ids := make([]string, 0, len(page.Items)) + pageIDs := make([]string, 0, len(page.Items)) + newItems := 0 + for i := range page.Items { + m := &page.Items[i] + pageIDs = append(pageIDs, m.ID) + if recent.contains(m.ID) { + continue + } + newItems++ + if persistsMessageRow(m) { + ids = append(ids, m.ID) + } + } + if newItems == 0 { + return false, nil + } + recent.add(pageIDs) + + if len(ids) > 0 { + archived, err := imp.store.ArchivedSourceMessageIDs(cc.sourceID, ids) + if err != nil { + sum.Errors++ + return false, nil //nolint:nilerr // a later scan retries this best-effort archive lookup + } + if ctxErr := ctx.Err(); ctxErr != nil { + return false, ctxErr + } + for _, id := range ids { + if _, ok := archived[id]; !ok { + cc.cs.Done = false + sum.ChatsReopened++ + return true, nil + } + } + return false, nil + } + + if !page.HasMore || page.OldestCursor == "" || page.OldestCursor == cursor { + return false, nil + } + cursor = page.OldestCursor + } + + // A very long event-only run is unusual. Route it through normal backfill + // rather than letting the probe bound hide content on every future scan. + cc.cs.Done = false + sum.ChatsReopened++ + return true, nil +} + // recentIDWindow remembers the message IDs of the last few pages of a // backfill walk. The live API's degenerate end-of-history pages re-serve the // immediately preceding tail, so a few pages of memory detect them — and stay @@ -599,7 +769,7 @@ func (imp *Importer) processMessage(ctx context.Context, cc *chatScope, m *Messa sum.MessagesProcessed++ return nil } - if m.IsHidden { + if !persistsMessageRow(m) { return nil } err := imp.persistMessage(ctx, cc, m, sum) @@ -609,6 +779,13 @@ func (imp *Importer) processMessage(ctx context.Context, cc *chatScope, m *Messa return err } +// persistsMessageRow reports whether processMessage archives a message row. +// Reactions update their target, deletions tombstone an existing row, and +// hidden events are intentionally omitted. +func persistsMessageRow(m *Message) bool { + return m.Type != "REACTION" && !m.IsDeleted && !m.IsHidden +} + // refreshReactionTarget re-fetches and re-persists the message a REACTION // event points at, refreshing its embedded reactions (and any edit). A 404 // target is expected churn; other fetch failures return errRetryPage so the @@ -827,6 +1004,17 @@ func (imp *Importer) recordItem(syncID int64, sourceMessageID, phase, status, ki }) } +// tailScanDue reports whether completed chats should be re-probed this run. +// An unset or unparseable timestamp counts as due, so archives written before +// tail scanning existed pick it up on their next sync. +func tailScanDue(last string, now time.Time) bool { + t := parseWatermark(last) + if t.IsZero() { + return true + } + return now.Sub(t) >= tailScanInterval +} + func parseWatermark(s string) time.Time { if s == "" { return time.Time{} diff --git a/internal/beeper/importer_test.go b/internal/beeper/importer_test.go index 5ac144015..f71fe73cd 100644 --- a/internal/beeper/importer_test.go +++ b/internal/beeper/importer_test.go @@ -3,6 +3,7 @@ package beeper import ( "context" "database/sql" + "slices" "strconv" "strings" "testing" @@ -714,6 +715,356 @@ func TestImportEmptyChatPicksUpLaterMessages(t *testing.T) { assert.Equal(1, total, "a chat that was empty at backfill must still pick up later messages") } +// TestImportPicksUpHistoryBackfilledAfterChatCompleted covers Beeper Desktop +// filling in a network's older history after msgvault already walked a chat to +// the end. The new messages carry old timestamps, so they neither advance the +// chat's lastActivity nor land in the reconcile window — without the tail scan +// the chat stays done and that history is never archived. +func TestImportPicksUpHistoryBackfilledAfterChatCompleted(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!grow:beeper.local", AccountID: "signal", Network: "Signal", Title: "Grow", Type: "single", + LastActivity: base.Add(2 * time.Minute), + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "n1", SortKey: 100, Timestamp: base, Text: "recent one", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + {ID: "n2", SortKey: 101, Timestamp: base.Add(time.Minute), Text: "recent two", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + var total int + require.NoError(st.DB().QueryRow(`SELECT COUNT(*) FROM messages WHERE message_type='beeper'`).Scan(&total)) + require.Equal(2, total, "first run archives the locally-available history") + + // The scan is daily in production; the second run below stands in for the + // next day's sync. + defer func(d time.Duration) { tailScanInterval = d }(tailScanInterval) + tailScanInterval = 0 + + // Beeper finishes its own backfill: older messages appear behind the ones + // already archived, without the chat's lastActivity changing. + f.prependMsgs("!grow:beeper.local", + fakeMsg{ID: "o1", SortKey: 1, Timestamp: base.Add(-48 * time.Hour), Text: "ancient one", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + fakeMsg{ID: "o2", SortKey: 2, Timestamp: base.Add(-47 * time.Hour), Text: "ancient two", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + ) + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + require.NoError(st.DB().QueryRow(`SELECT COUNT(*) FROM messages WHERE message_type='beeper'`).Scan(&total)) + assert.Equal(4, total, "history Beeper backfilled after the chat completed must be archived") + assert.Equal(int64(1), sum.ChatsReopened) + + var text string + require.NoError(st.DB().QueryRow( + `SELECT b.body_text FROM messages m JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'o1'`).Scan(&text)) + assert.Equal("ancient one", text) +} + +func TestImportTailOnlyCursorlessChatStillChecksForBackfill(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + recent := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!empty-tail:beeper.local", AccountID: "signal", Network: "Signal", Title: "Empty", Type: "single", + LastActivity: base, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + }) + f.addChat(&fakeChat{ + ID: "!active-empty-tail:beeper.local", AccountID: "signal", Network: "Signal", Title: "Active", Type: "single", + LastActivity: recent, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "a1", SortKey: 1, Timestamp: recent, Text: "active", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + oldInterval := tailScanInterval + tailScanInterval = 0 + t.Cleanup(func() { tailScanInterval = oldInterval }) + + // Beeper later fills an initially empty chat without advancing its listed + // activity. The newer chat makes this one eligible only through tail scan. + f.prependMsgs("!empty-tail:beeper.local", fakeMsg{ + ID: "late-old", SortKey: 1, Timestamp: base.Add(-time.Hour), Text: "arrived later", + SenderID: "@signal_ann:beeper.local", SenderName: "Ann", + }) + + _, err = imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + var total int + require.NoError(st.DB().QueryRow(`SELECT COUNT(*) FROM messages WHERE message_type='beeper'`).Scan(&total)) + assert.Equal(2, total, "a cursorless tail-only chat must still run its empty-chat recovery") +} + +func TestImportTailProbeSkipsEventOnlyPagesToFindOlderContent(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + recent := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.pageSize = 2 + f.addChat(&fakeChat{ + ID: "!event-page:beeper.local", AccountID: "signal", Network: "Signal", Title: "Events", Type: "single", + LastActivity: base, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "q1", SortKey: 100, Timestamp: base, Text: "known", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + f.addChat(&fakeChat{ + ID: "!active-event-page:beeper.local", AccountID: "signal", Network: "Signal", Title: "Active", Type: "single", + LastActivity: recent, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "a1", SortKey: 100, Timestamp: recent, Text: "active", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + oldInterval := tailScanInterval + tailScanInterval = 0 + t.Cleanup(func() { tailScanInterval = oldInterval }) + + // The first page behind the stored cursor contains no row-producing + // messages; the older content is visible only after following its cursor. + f.prependMsgs("!event-page:beeper.local", + fakeMsg{ID: "older-content", SortKey: 1, Timestamp: base.Add(-3 * time.Hour), Text: "older content", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + fakeMsg{ID: "hidden-between", SortKey: 2, Timestamp: base.Add(-2 * time.Hour), IsHidden: true}, + fakeMsg{ID: "reaction-between", SortKey: 3, Timestamp: base.Add(-time.Hour), Type: "REACTION", IsHidden: true, LinkedMessageID: "q1"}, + ) + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Equal(int64(1), sum.ChatsReopened) + + var text string + require.NoError(st.DB().QueryRow( + `SELECT b.body_text FROM messages m JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'older-content'`).Scan(&text)) + assert.Equal("older content", text) +} + +func TestImportTailScanIgnoresUnarchivedNonContentEvents(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + recent := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!quiet-events:beeper.local", AccountID: "signal", Network: "Signal", Title: "Quiet", Type: "single", + LastActivity: base, TailCountdown: true, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "q1", SortKey: 100, Timestamp: base, Text: "hello", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + f.addChat(&fakeChat{ + ID: "!active-events:beeper.local", AccountID: "signal", Network: "Signal", Title: "Active", Type: "single", + LastActivity: recent, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "a1", SortKey: 100, Timestamp: recent, Text: "active", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + oldInterval := tailScanInterval + tailScanInterval = 0 + t.Cleanup(func() { tailScanInterval = oldInterval }) + + // These events sit behind the completed cursor but deliberately create no + // message rows. A degenerate tail page will keep re-serving them, so treating + // their absent IDs as archive gaps would reopen the chat on every scan. + f.prependMsgs("!quiet-events:beeper.local", + fakeMsg{ID: "reaction-old", SortKey: 1, Timestamp: base.Add(-3 * time.Hour), Type: "REACTION", IsHidden: true, LinkedMessageID: "q1"}, + fakeMsg{ID: "hidden-old", SortKey: 2, Timestamp: base.Add(-2 * time.Hour), Text: "hidden", IsHidden: true}, + fakeMsg{ID: "deleted-old", SortKey: 3, Timestamp: base.Add(-time.Hour), IsDeleted: true}, + ) + + for range 2 { + sum, ierr := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(ierr) + assert.Zero(sum.ChatsReopened, "non-content events must not reopen a completed chat") + } + + var total int + require.NoError(st.DB().QueryRow(`SELECT COUNT(*) FROM messages WHERE message_type='beeper'`).Scan(&total)) + assert.Equal(2, total, "non-content events must remain excluded from the archive") +} + +func TestImportTailOnlyUnchangedChatSkipsIncrementalAndReconcile(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + recent := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!quiet-probe:beeper.local", AccountID: "signal", Network: "Signal", Title: "Quiet", Type: "single", + LastActivity: base, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "q1", SortKey: 1, Timestamp: base, Text: "quiet", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + f.addChat(&fakeChat{ + ID: "!active-probe:beeper.local", AccountID: "signal", Network: "Signal", Title: "Active", Type: "single", + LastActivity: recent, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "a1", SortKey: 1, Timestamp: recent, Text: "active", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, _, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + oldInterval := tailScanInterval + tailScanInterval = 0 + t.Cleanup(func() { tailScanInterval = oldInterval }) + f.resetRequests() + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Zero(sum.ChatsReopened) + + var quietRequests []string + for _, req := range f.requests() { + if strings.Contains(req, "/v1/chats/!quiet-probe:beeper.local/messages") && + !strings.Contains(req, "/messages/") { + quietRequests = append(quietRequests, req) + } + } + require.Len(quietRequests, 1, "an unchanged tail-only chat needs only its tail probe") + assert.Contains(quietRequests[0], "direction=before") +} + +func TestImportTailProbeCancellationLeavesScanDue(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + recent := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!active-cancel-probe:beeper.local", AccountID: "signal", Network: "Signal", Title: "Active", Type: "single", + LastActivity: recent, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "active-cancel", SortKey: 1, Timestamp: recent, Text: "active", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + f.addChat(&fakeChat{ + ID: "!quiet-cancel-probe:beeper.local", AccountID: "signal", Network: "Signal", Title: "Quiet", Type: "single", + LastActivity: base, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "quiet-cancel", SortKey: 1, Timestamp: base, Text: "quiet", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + src, err := st.GetOrCreateSource("beeper", "signal") + require.NoError(err) + run, err := st.GetLastSuccessfulSync(src.ID) + require.NoError(err) + require.True(run.CursorAfter.Valid) + state, err := LoadSyncState(run.CursorAfter.String) + require.NoError(err) + state.LastTailScan = formatWatermark(time.Now().Add(-25 * time.Hour)) + blob, err := state.Marshal() + require.NoError(err) + _, err = st.DB().Exec(st.Rebind(`UPDATE sync_runs SET cursor_after = ? WHERE id = ?`), blob, run.ID) + require.NoError(err) + + ctx, cancel := context.WithCancel(context.Background()) + f.cancelMessageListFor("!quiet-cancel-probe:beeper.local", cancel) + _, err = imp.Import(ctx, ImportOptions{AccountID: "signal"}) + require.ErrorIs(err, context.Canceled) + + // The interrupted probe must leave the old stamp in the failed checkpoint, + // so an immediate retry still visits the otherwise-inactive quiet chat. + f.resetRequests() + _, err = imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.True(slices.ContainsFunc(f.requests(), func(req string) bool { + return strings.Contains(req, "/v1/chats/!quiet-cancel-probe:beeper.local/messages") && + strings.Contains(req, "direction=before") + }), "an interrupted tail scan must remain due for immediate retry") +} + +// TestImportTailScanThrottled covers the probe not running on every sync: it +// costs one request per completed chat, so a run soon after a clean scan must +// leave completed chats alone. +func TestImportTailScanThrottled(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + f := newFakeBeeper(t) + f.addChat(&fakeChat{ + ID: "!quiet:beeper.local", AccountID: "signal", Network: "Signal", Title: "Quiet", Type: "single", + LastActivity: base, + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + {ID: "q1", SortKey: 1, Timestamp: base, Text: "hello", SenderID: "@signal_ann:beeper.local", SenderName: "Ann"}, + }, + }) + imp, _, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Zero(sum.ChatsReopened, "a scan within the interval must not re-probe completed chats") +} + +func TestTailScanDue(t *testing.T) { + assert := assert.New(t) + now := time.Date(2026, 8, 8, 12, 0, 0, 0, time.UTC) + assert.True(tailScanDue("", now), "an archive predating tail scanning is due") + assert.True(tailScanDue("not-a-timestamp", now), "an unparseable stamp is due") + assert.True(tailScanDue(formatWatermark(now.Add(-25*time.Hour)), now)) + assert.False(tailScanDue(formatWatermark(now.Add(-time.Hour)), now)) +} + func TestImportDegenerateTailPagination(t *testing.T) { require := require.New(t) assert := assert.New(t) diff --git a/internal/beeper/mapping.go b/internal/beeper/mapping.go index 907083b40..d01795324 100644 --- a/internal/beeper/mapping.go +++ b/internal/beeper/mapping.go @@ -2,9 +2,13 @@ package beeper import ( "database/sql" + "html" + "regexp" "strings" + "go.kenn.io/msgvault/internal/mime" "go.kenn.io/msgvault/internal/store" + "go.kenn.io/msgvault/internal/textutil" ) // messageType is the msgvault message_type for all Beeper-archived messages. @@ -13,6 +17,47 @@ import ( // while message_type values live in fixed lists across the query layer. const messageType = "beeper" +// htmlElementRe matches an opening or closing tag of an HTML element Beeper +// actually emits in message text. Matching a known element name (rather than +// any "<...>") keeps ordinary prose containing angle brackets — "a < b", "<3", +// "see below" — from being mistaken for markup and mangled. +var htmlElementRe = regexp.MustCompile(`(?i)]*)?/?>`) + +// mxReplyFallbackRe removes Matrix's quoted reply fallback before generic HTML +// conversion. The block repeats the parent message for clients that do not +// understand reply relations; keeping it would index the parent as new text. +var mxReplyFallbackRe = regexp.MustCompile(`(?is)]*)?>.*?`) + +// htmlEntityRe matches the named and numeric entities that show up in +// otherwise tag-free text (e.g. "at 2:15&k?"). A bare "&" is left alone, +// so plain messages like "Help & About" are never rewritten. +var htmlEntityRe = regexp.MustCompile(`&(amp|lt|gt|quot|apos|nbsp|#\d+|#x[0-9a-fA-F]+);`) + +// plainText renders a Beeper message's text as plain text. +// +// The API's `text` field is HTML for a large minority of messages (formatted +// Matrix messages, Telegram custom emoji, link previews) and genuinely plain +// for the rest, with no field distinguishing the two — so the shape of the +// value has to be detected before converting. Storing it verbatim puts markup +// into message bodies, snippets and the FTS index, where `target="_blank"` +// swamps searches for the ordinary word "target". +// +// The archived raw JSON (raw_format = beeper_json) keeps the original HTML, so +// this conversion is never lossy for the archive itself. +func plainText(s string) string { + s = mxReplyFallbackRe.ReplaceAllString(s, "") + var text string + switch { + case htmlElementRe.MatchString(s): + text = mime.StripHTML(s) + case htmlEntityRe.MatchString(s): + text = html.UnescapeString(s) + default: + text = s + } + return textutil.SanitizeTerminalMultiline(text) +} + func snippet(text string) string { r := []rune(text) if len(r) > 100 { @@ -21,11 +66,15 @@ func snippet(text string) string { return text } +// typeImage is the Beeper message type for a photo — including the link +// previews that arrive typed as images rather than as the media they preview. +const typeImage = "IMAGE" + // placeholderBody synthesizes a searchable body line for messages that carry // no text (media, stickers, locations). func placeholderBody(m *Message) string { switch m.Type { - case "IMAGE": + case typeImage: return "[image]" case "VIDEO": return "[video]" @@ -54,7 +103,7 @@ func placeholderBody(m *Message) string { // visible to FTS and embeddings. func bodyText(m *Message) string { var parts []string - if text := strings.TrimSpace(m.Text); text != "" { + if text := strings.TrimSpace(plainText(m.Text)); text != "" { parts = append(parts, text) } else if ph := placeholderBody(m); ph != "" { parts = append(parts, ph) @@ -67,6 +116,30 @@ func bodyText(m *Message) string { return strings.Join(parts, "\n") } +// urlOnlyRe matches a body consisting of exactly one URL and nothing else. +var urlOnlyRe = regexp.MustCompile(`^https?://\S+$`) + +// sharedLink reports the URL a message is forwarding, when the message is a +// link share rather than something the sender composed: its whole body is a +// single URL and it carries an attachment, which is the shape every network +// uses for a link preview (an Instagram reel, an x.com post, a GitHub link). +// +// The distinction matters for storage accounting. A forwarded reel and a photo +// a friend took are both media rows, but the reel's bytes are a preview of +// somebody else's public post — recoverable from the URL — while the photo is +// not. Recording the share URL lets those be told apart after the fact without +// re-reading every archived blob. +func sharedLink(m *Message) string { + if len(m.Attachments) == 0 { + return "" + } + text := strings.TrimSpace(plainText(m.Text)) + if !urlOnlyRe.MatchString(text) { + return "" + } + return text +} + // mapMessage converts a Beeper API Message into a store.Message plus the // plain-text body. conversationID and sourceID are internal DB IDs. The // message's own numeric id is the source_message_id (unique per installation; diff --git a/internal/beeper/mapping_test.go b/internal/beeper/mapping_test.go index b005bdac4..c31aed889 100644 --- a/internal/beeper/mapping_test.go +++ b/internal/beeper/mapping_test.go @@ -16,7 +16,7 @@ func TestBodyText(t *testing.T) { want string }{ {"plain text", Message{Type: "TEXT", Text: "hello"}, "hello"}, - {"image placeholder", Message{Type: "IMAGE"}, "[image]"}, + {"image placeholder", Message{Type: typeImage}, "[image]"}, {"video placeholder", Message{Type: "VIDEO"}, "[video]"}, {"voice placeholder", Message{Type: "VOICE"}, "[voice message]"}, {"sticker placeholder", Message{Type: "STICKER"}, "[sticker]"}, @@ -24,7 +24,7 @@ func TestBodyText(t *testing.T) { {"location with text keeps text", Message{Type: "LOCATION", Text: "123 Main St"}, "123 Main St"}, {"file with name", Message{Type: "FILE", Attachments: []Attachment{{FileName: "report.pdf"}}}, "[file: report.pdf]"}, {"file without name", Message{Type: "FILE"}, "[file]"}, - {"media with caption keeps caption", Message{Type: "IMAGE", Text: "look at this"}, "look at this"}, + {"media with caption keeps caption", Message{Type: typeImage, Text: "look at this"}, "look at this"}, { "voice transcription appended", Message{Type: "VOICE", Attachments: []Attachment{{IsVoiceNote: true, Transcription: &Transcription{Transcription: "call me back"}}}}, @@ -38,6 +38,118 @@ func TestBodyText(t *testing.T) { } } +// TestBodyTextHTML covers the Beeper API serving HTML in the `text` field for +// some messages and plain text for others, with no field distinguishing them. +// The inputs are shapes observed from a live account across bridges. +func TestBodyTextHTML(t *testing.T) { + tests := []struct { + name string + text string + want string + }{ + { + "matrix formatted message with mention link", + `

@bob, good to know.

`, + "@bob, good to know.", + }, + { + "matrix reply fallback is excluded", + `
In reply to Alice
quoted parent text

fresh reply

`, + "fresh reply", + }, + { + "whatsapp line breaks", + "Thanks :)

Either way we will be around.", + "Thanks :)\n\nEither way we will be around.", + }, + { + "entity-only text is unescaped without tag stripping", + "meet at 2:15&k?", + "meet at 2:15&k?", + }, + { + "numeric control entities cannot inject terminal escapes", + "first line\nsecond line]52;c;evilsafe›tail", + "first line\nsecond linesafe›tail", + }, + { + "telegram custom emoji image is dropped", + `written in go too `, + "written in go too", + }, + { + "instagram share collapses to the bare url", + `https://www.instagram.com/p/ABC123/`, + "https://www.instagram.com/p/ABC123/", + }, + // Plain messages must survive untouched: angle brackets and bare + // ampersands are ordinary prose, not markup. + {"bare ampersand is left alone", "Settings, Help & About", "Settings, Help & About"}, + {"comparison is not a tag", "confirmed a < b for all inputs", "confirmed a < b for all inputs"}, + {"emoticon is not a tag", "nice work <3", "nice work <3"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, bodyText(&Message{Type: "TEXT", Text: tt.text})) + }) + } +} + +func TestMapMessageDropsMatrixReplyFallbackFromSnippet(t *testing.T) { + formatted := `
quoted parent text

fresh reply

` + msg, body := mapMessage(&Message{ID: "reply", Type: "TEXT", Text: formatted}, 1, 1) + assert.Equal(t, "fresh reply", body) + assert.Equal(t, "fresh reply", msg.Snippet.String) +} + +// TestSharedLink covers telling a forwarded link preview apart from media the +// sender composed: the share's whole body is the URL it previews. +func TestSharedLink(t *testing.T) { + att := []Attachment{{FileName: "image.jpg"}} + tests := []struct { + name string + msg Message + want string + }{ + { + "instagram reel share", + Message{Type: typeImage, Text: `https://www.instagram.com/p/ABC/`, Attachments: att}, + "https://www.instagram.com/p/ABC/", + }, + { + "plain url share", + Message{Type: typeImage, Text: "https://github.com/example/repo", Attachments: att}, + "https://github.com/example/repo", + }, + {"photo with no text is not a share", Message{Type: typeImage, Attachments: att}, ""}, + { + "url with commentary is the sender's own message", + Message{Type: typeImage, Text: "look at this https://www.instagram.com/p/ABC/", Attachments: att}, + "", + }, + {"link without media is not a share", Message{Type: "TEXT", Text: "https://example.com"}, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, sharedLink(&tt.msg)) + }) + } +} + +func TestShareMetadata(t *testing.T) { + assert := assert.New(t) + att := []Attachment{{FileName: "image.jpg"}} + assert.Empty(shareMetadata(&Message{Type: typeImage, Attachments: att})) + assert.JSONEq( + `{"shared_url":"https://www.instagram.com/p/ABC/"}`, + shareMetadata(&Message{Type: typeImage, Text: "https://www.instagram.com/p/ABC/", Attachments: att}), + ) + // A URL carrying a quote must not be able to break out of the JSON value. + meta := shareMetadata(&Message{Type: typeImage, Text: `https://x.example/"+evil`, Attachments: att}) + assert.NotEmpty(meta) + assert.JSONEq(`{"shared_url":"https://x.example/\"+evil"}`, meta) +} + func TestMapMessage(t *testing.T) { require := require.New(t) assert := assert.New(t) diff --git a/internal/beeper/media.go b/internal/beeper/media.go index b18f71a74..aa300e2fc 100644 --- a/internal/beeper/media.go +++ b/internal/beeper/media.go @@ -2,6 +2,7 @@ package beeper import ( "context" + "encoding/json" "errors" "fmt" "time" @@ -61,6 +62,25 @@ func declaredSize(att *Attachment) int { return int(int64(att.FileSize)) } +// shareMetadata renders the attachment_metadata JSON marking media that came +// in as a link preview rather than as something the sender composed, recording +// the URL it previews. Returns "" for ordinary media, which stores NULL. +func shareMetadata(m *Message) string { + link := sharedLink(m) + if link == "" { + return "" + } + // Marshal rather than concatenate: the URL is untrusted remote input and + // must not be able to break out of the JSON value. + b, err := json.Marshal(struct { + SharedURL string `json:"shared_url"` + }{SharedURL: link}) + if err != nil { + return "" + } + return string(b) +} + // persistAttachments downloads a message's media into content-addressed // storage and replaces the message's Beeper attachment rows. Media already // downloaded for this message (matched by source_attachment_id) is kept @@ -84,6 +104,7 @@ func (imp *Importer) persistAttachments(ctx context.Context, syncID, messageID i if maxBytes <= 0 { maxBytes = defaultMaxMediaBytes } + shareMeta := shareMetadata(m) refs := make([]store.AttachmentRef, 0, len(m.Attachments)) for i := range m.Attachments { att := &m.Attachments[i] @@ -93,6 +114,10 @@ func (imp *Importer) persistAttachments(ctx context.Context, syncID, messageID i } sourceAttID := beeperAttachmentID(ref) if prev, ok := existing[sourceAttID]; ok && prev.ContentHash != "" { + // Re-persisting already-downloaded media: keep the blob as-is but + // refresh the share marker, so re-running over an existing archive + // classifies rows stored before this was recorded. + prev.Metadata = shareMeta refs = append(refs, prev) continue } @@ -102,6 +127,7 @@ func (imp *Importer) persistAttachments(ctx context.Context, syncID, messageID i StoragePath: ref, Size: declaredSize(att), SourceAttachmentID: sourceAttID, + Metadata: shareMeta, } // Every failure leaves the marker as a pending row (BackfillMedia // retries it); only unexpected failures also count as errors. @@ -147,6 +173,7 @@ func (imp *Importer) persistAttachments(ctx context.Context, syncID, messageID i SourceAttachmentID: sourceAttID, MediaType: mediaTypeOf(att), DurationMS: int64(att.Duration * 1000), + Metadata: shareMeta, } if att.Size != nil { stored.Width = int64(att.Size.Width) diff --git a/internal/beeper/repair.go b/internal/beeper/repair.go new file mode 100644 index 000000000..ebabf00ce --- /dev/null +++ b/internal/beeper/repair.go @@ -0,0 +1,131 @@ +package beeper + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "time" + + "go.kenn.io/msgvault/internal/rederive" + "go.kenn.io/msgvault/internal/store" +) + +// rawArchiveFormat is the message_raw.raw_format tag Beeper messages archive +// under; the repair reads back exactly what the importer wrote. +const rawArchiveFormat = "beeper_json" + +// rederiveVersion identifies this package's derivation logic. Bump it whenever +// a change would produce different body text, snippets or attachment metadata +// for the same payload, so existing archives re-derive on their next sync. +// +// v1 — plain-text conversion of HTML message text; link-preview classification. +// v2 — refresh snippets and the search index even when body text is current. +// v3 — exclude Matrix reply fallbacks from body text, snippets, and search. +const rederiveVersion = "v3" + +// repairBatchSize bounds how many archived messages are held in memory per +// pass of the walk. +const repairBatchSize = 500 + +func init() { + rederive.Register(sourceTypeBeeper, rederiveVersion, + func(ctx context.Context, s *store.Store, sourceID int64, progress func(string)) (*rederive.Summary, error) { + // The pass never calls Beeper, so a nil client is correct here. + return NewImporter(s, nil).RepairSource(ctx, sourceID, progress) + }) +} + +// RepairSource re-derives stored message text and attachment classification for +// one Beeper source from the verbatim JSON archived alongside each message. +// +// Both are computed from the API payload at import time, so changing how they +// are derived leaves already-archived rows stale. Reading the archive rather +// than re-fetching keeps the pass offline, fast, and able to repair messages +// Beeper Desktop no longer holds — a re-sync could do neither. +// +// Only derived columns are rewritten: message bodies, snippets, the search +// index, and attachment metadata. Raw payloads, stored media, and sync cursors +// are left alone, so the pass is idempotent and safe to re-run. It touches no +// Beeper endpoint, so an Importer built with a nil client is valid. +func (imp *Importer) RepairSource(ctx context.Context, sourceID int64, progress func(string)) (*rederive.Summary, error) { + start := time.Now() + sum := &rederive.Summary{} + + var afterID int64 + for { + if err := ctx.Err(); err != nil { + return sum, err + } + batch, err := imp.store.ScanArchivedRawMessages(sourceID, rawArchiveFormat, afterID, repairBatchSize) + if err != nil { + return sum, err + } + if len(batch) == 0 { + break + } + for i := range batch { + if err := ctx.Err(); err != nil { + return sum, err + } + item := &batch[i] + afterID = item.MessageID + sum.MessagesScanned++ + imp.repairMessage(item, sourceID, sum) + } + if progress != nil { + progress(fmt.Sprintf("%d scanned, %d bodies rewritten, %d attachments tagged", + sum.MessagesScanned, sum.BodiesRewritten, sum.AttachmentsTagged)) + } + } + sum.Duration = time.Since(start) + return sum, nil +} + +// repairMessage re-derives one archived message's stored text and attachment +// classification. Per-message failures are counted rather than fatal: one +// unreadable payload must not abandon the rest of the archive. +func (imp *Importer) repairMessage(item *store.ArchivedRawMessage, sourceID int64, sum *rederive.Summary) { + var m Message + if err := json.Unmarshal(item.RawData, &m); err != nil { + sum.Undecodable++ + return + } + // Reactions, tombstones, and hidden events never had a body of their own; + // the importer skips them at persist time, so there is nothing to re-derive. + if !persistsMessageRow(&m) { + return + } + + // Re-run the production mapping so repaired rows are exactly what a fresh + // sync would now write. + msg, text := mapMessage(&m, item.ConversationID, sourceID) + + // Refresh the whole derived-text tuple even when the body is current. + // Snippet and FTS state can drift independently, and the store commits all + // three fields atomically so a failure leaves the row retryable. + bodyChanged := item.BodyText != text + if err := imp.store.UpdateMessageDerivedText( + item.MessageID, + sql.NullString{String: text, Valid: text != ""}, + sql.NullString{}, + msg.Snippet, + store.FTSDoc{Body: text, FromAddr: m.SenderName}, + ); err != nil { + sum.Errors++ + return + } + if bodyChanged { + sum.BodiesRewritten++ + } + + if len(m.Attachments) == 0 { + return + } + changed, err := imp.store.SetBeeperAttachmentMetadata(item.MessageID, shareMetadata(&m)) + if err != nil { + sum.Errors++ + return + } + sum.AttachmentsTagged += changed +} diff --git a/internal/beeper/repair_test.go b/internal/beeper/repair_test.go new file mode 100644 index 000000000..2ed66e71e --- /dev/null +++ b/internal/beeper/repair_test.go @@ -0,0 +1,362 @@ +package beeper + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/msgvault/internal/store" + "go.kenn.io/msgvault/internal/testutil" +) + +// TestRepairArchiveRewritesStaleDerivedRows covers repairing an archive +// written before HTML was converted and before shares were classified: the +// pass must reach both from the stored payload alone, without contacting +// Beeper Desktop. +func TestRepairArchiveRewritesStaleDerivedRows(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + ch := shareAndHTMLChat() + f.addChat(ch) + f.setAsset("mxc://x/share1", []byte("share-preview-bytes")) + f.setAsset("mxc://x/photo1", []byte("photo-bytes")) + + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal", AttachmentsDir: t.TempDir()}) + require.NoError(err) + + // Simulate rows written by an older build: raw HTML in the body and no + // share classification on the attachments. + _, err = st.DB().Exec(st.Rebind(`UPDATE message_bodies SET body_text = ? + WHERE message_id = (SELECT id FROM messages WHERE source_message_id = 'html1')`), + `

hello there

`) + require.NoError(err) + _, err = st.DB().Exec(`UPDATE attachments SET attachment_metadata = NULL`) + require.NoError(err) + + sum, err := imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + assert.Zero(sum.Errors) + assert.Zero(sum.Undecodable) + assert.Equal(int64(1), sum.BodiesRewritten, "only the stale body needs rewriting") + + var body, snippet string + require.NoError(st.DB().QueryRow(`SELECT b.body_text, COALESCE(m.snippet, '') + FROM messages m JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'html1'`).Scan(&body, &snippet)) + assert.Equal("hello there", body, "HTML must be converted to plain text") + assert.Equal("hello there", snippet) + + // The forwarded link keeps its share URL; the photo stays unclassified. + var shareMeta, photoMeta string + require.NoError(st.DB().QueryRow(`SELECT COALESCE(CAST(a.attachment_metadata AS TEXT), '') + FROM attachments a JOIN messages m ON m.id = a.message_id + WHERE m.source_message_id = 'share1'`).Scan(&shareMeta)) + require.NoError(st.DB().QueryRow(`SELECT COALESCE(CAST(a.attachment_metadata AS TEXT), '') + FROM attachments a JOIN messages m ON m.id = a.message_id + WHERE m.source_message_id = 'photo1'`).Scan(&photoMeta)) + assert.JSONEq(`{"shared_url":"https://www.instagram.com/p/ABC/"}`, shareMeta) + assert.Empty(photoMeta, "media the sender composed is not a share") + + // Re-running must be a no-op: nothing left differing from the archive. + again, err := imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + assert.Zero(again.BodiesRewritten) + assert.Zero(again.AttachmentsTagged) + assert.Equal(sum.MessagesScanned, again.MessagesScanned) +} + +func TestRepairArchiveRollsBackDerivedTextTogether(t *testing.T) { + testutil.SkipIfPostgres(t, "SQLite trigger injects a failure after the body write") + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + f.addChat(shareAndHTMLChat()) + f.setAsset("mxc://x/share1", []byte("share-preview-bytes")) + f.setAsset("mxc://x/photo1", []byte("photo-bytes")) + + imp, st, done := newTestImporter(t, f) + defer done() + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal", AttachmentsDir: t.TempDir()}) + require.NoError(err) + + const stale = `

hello there

` + _, err = st.DB().Exec(st.Rebind(`UPDATE message_bodies SET body_text = ? + WHERE message_id = (SELECT id FROM messages WHERE source_message_id = 'html1')`), stale) + require.NoError(err) + _, err = st.DB().Exec(`CREATE TRIGGER fail_repair_snippet + BEFORE UPDATE OF snippet ON messages + WHEN OLD.source_message_id = 'html1' + BEGIN SELECT RAISE(ABORT, 'injected snippet failure'); END`) + require.NoError(err) + + sum, err := imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + assert.Equal(int64(1), sum.Errors) + + var body string + require.NoError(st.DB().QueryRow(`SELECT b.body_text FROM messages m + JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'html1'`).Scan(&body)) + assert.Equal(stale, body, "a later derived-field failure must roll back the body write") + + _, err = st.DB().Exec(`DROP TRIGGER fail_repair_snippet`) + require.NoError(err) + sum, err = imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + assert.Equal(int64(1), sum.BodiesRewritten) + require.NoError(st.DB().QueryRow(`SELECT b.body_text FROM messages m + JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'html1'`).Scan(&body)) + assert.Equal("hello there", body) +} + +func TestRepairArchiveRefreshesSnippetAndFTSWhenBodyIsCurrent(t *testing.T) { + testutil.SkipIfPostgres(t, "directly corrupts the SQLite FTS5 table") + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + f.addChat(shareAndHTMLChat()) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + require.True(st.FTS5Available(), "FTS5 must be available for the repair regression") + + var messageID int64 + require.NoError(st.DB().QueryRow( + `SELECT id FROM messages WHERE source_message_id = 'html1'`).Scan(&messageID)) + _, err = st.DB().Exec(`UPDATE messages SET snippet = 'stale snippet' WHERE id = ?`, messageID) + require.NoError(err) + _, err = st.DB().Exec(`DELETE FROM messages_fts WHERE rowid = ?`, messageID) + require.NoError(err) + + sum, err := imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + assert.Zero(sum.Errors) + assert.Zero(sum.BodiesRewritten, "the body was already current") + + var body, snippet string + require.NoError(st.DB().QueryRow(`SELECT b.body_text, COALESCE(m.snippet, '') + FROM messages m JOIN message_bodies b ON b.message_id = m.id + WHERE m.id = ?`, messageID).Scan(&body, &snippet)) + assert.Equal("hello there", body) + assert.Equal("hello there", snippet, "repair must refresh a stale snippet independently of body drift") + + var ftsHits int + require.NoError(st.DB().QueryRow( + `SELECT COUNT(*) FROM messages_fts WHERE rowid = ? AND messages_fts MATCH 'hello'`, messageID).Scan(&ftsHits)) + assert.Equal(1, ftsHits, "repair must restore a missing FTS row independently of body drift") +} + +func TestSyncReportsIncompleteRepair(t *testing.T) { + testutil.SkipIfPostgres(t, "SQLite trigger injects a row-level metadata failure") + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + f.addChat(shareAndHTMLChat()) + f.setAsset("mxc://x/share1", []byte("share-preview-bytes")) + f.setAsset("mxc://x/photo1", []byte("photo-bytes")) + + imp, st, done := newTestImporter(t, f) + defer done() + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal", AttachmentsDir: t.TempDir()}) + require.NoError(err) + _, err = st.DB().Exec(`DELETE FROM applied_migrations WHERE name LIKE 'rederive:%'`) + require.NoError(err) + _, err = st.DB().Exec(`UPDATE attachments SET attachment_metadata = NULL`) + require.NoError(err) + _, err = st.DB().Exec(`CREATE TRIGGER fail_repair_metadata + BEFORE UPDATE OF attachment_metadata ON attachments + BEGIN SELECT RAISE(ABORT, 'injected metadata failure'); END`) + require.NoError(err) + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Equal(int64(1), sum.Errors, "row-level repair failures must reach the sync summary") +} + +// TestRepairArchiveLeavesStoredMediaAlone covers the pass rewriting only +// derived columns: a repair must never disturb downloaded blobs. +func TestRepairArchiveLeavesStoredMediaAlone(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + f.addChat(shareAndHTMLChat()) + f.setAsset("mxc://x/share1", []byte("share-preview-bytes")) + f.setAsset("mxc://x/photo1", []byte("photo-bytes")) + + imp, st, done := newTestImporter(t, f) + defer done() + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal", AttachmentsDir: t.TempDir()}) + require.NoError(err) + + before := storedBlobs(t, st) + require.NotEmpty(before) + + _, err = imp.RepairSource(context.Background(), beeperSourceID(t, st), nil) + require.NoError(err) + + assert.Equal(before, storedBlobs(t, st), "repair must not touch stored blobs") +} + +// storedBlobs maps each attachment to the blob it points at, so a test can +// assert that a pass left stored media untouched. +func storedBlobs(t *testing.T, st *store.Store) map[string]string { + t.Helper() + require := require.New(t) + rows, err := st.DB().Query( + `SELECT source_attachment_id, COALESCE(content_hash, '') || '|' || storage_path FROM attachments`) + require.NoError(err) + defer func() { _ = rows.Close() }() + out := map[string]string{} + for rows.Next() { + var k, v string + require.NoError(rows.Scan(&k, &v)) + out[k] = v + } + require.NoError(rows.Err()) + return out +} + +// shareAndHTMLChat builds a chat holding one HTML message, one forwarded link +// preview, and one photo the sender composed. +func shareAndHTMLChat() *fakeChat { + // Older than the reconcile window so head re-walks terminate immediately. + base := time.Now().Add(-60 * 24 * time.Hour).UTC().Truncate(time.Second) + return &fakeChat{ + ID: "!repair:beeper.local", AccountID: "signal", Network: "Signal", + Title: "Repair", Type: "single", LastActivity: base.Add(2 * time.Minute), + Participants: []map[string]any{{"id": "@me:beeper.local", "isSelf": true}}, + Msgs: []fakeMsg{ + { + ID: "html1", SortKey: 1, Timestamp: base, + Text: `

hello there

`, + SenderID: "@signal_ann:beeper.local", SenderName: "Ann", + }, + { + ID: "share1", SortKey: 2, Timestamp: base.Add(time.Minute), Type: typeImage, + Text: `https://www.instagram.com/p/ABC/`, + SenderID: "@signal_ann:beeper.local", SenderName: "Ann", + Attachments: []map[string]any{{"id": "mxc://x/share1", "type": "img", "mimeType": "image/jpeg"}}, + }, + { + ID: "photo1", SortKey: 3, Timestamp: base.Add(2 * time.Minute), Type: typeImage, + SenderID: "@signal_ann:beeper.local", SenderName: "Ann", + Attachments: []map[string]any{{"id": "mxc://x/photo1", "type": "img", "mimeType": "image/jpeg"}}, + }, + }, + } +} + +// TestSyncHealsRowsFromAnOlderBuild covers an upgraded archive converging on +// its own: the next sync re-derives rows written before the derivation changed, +// without the user knowing a repair exists. The ledger then keeps later syncs +// from repeating the work. +func TestSyncHealsRowsFromAnOlderBuild(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + f.addChat(shareAndHTMLChat()) + f.setAsset("mxc://x/share1", []byte("share-preview-bytes")) + f.setAsset("mxc://x/photo1", []byte("photo-bytes")) + + imp, st, done := newTestImporter(t, f) + defer done() + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal", AttachmentsDir: t.TempDir()}) + require.NoError(err) + + // Put the archive back into the state an older build would have left, and + // clear the ledger so the source looks un-repaired. + _, err = st.DB().Exec(st.Rebind(`UPDATE message_bodies SET body_text = ? + WHERE message_id = (SELECT id FROM messages WHERE source_message_id = 'html1')`), + `

hello there

`) + require.NoError(err) + _, err = st.DB().Exec(`DELETE FROM applied_migrations WHERE name LIKE 'rederive:%'`) + require.NoError(err) + + sum, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Equal(int64(1), sum.BodiesRepaired, "the sync must heal the stale row") + + var body string + require.NoError(st.DB().QueryRow(`SELECT b.body_text FROM messages m + JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_message_id = 'html1'`).Scan(&body)) + assert.Equal("hello there", body) + + // The ledger now records the pass, so the next sync must not redo it. + next, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + assert.Zero(next.BodiesRepaired, "a healed archive must not re-derive on every sync") +} + +func TestSyncRunsCurrentRepairAfterV2WasApplied(t *testing.T) { + testutil.SkipIfPostgres(t, "directly corrupts the SQLite FTS5 table") + require := require.New(t) + assert := assert.New(t) + + f := newFakeBeeper(t) + ch := shareAndHTMLChat() + ch.Msgs[0].Text = `
quoted parent text

fresh reply

` + f.addChat(ch) + imp, st, done := newTestImporter(t, f) + defer done() + + _, err := imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + require.True(st.FTS5Available(), "FTS5 must be available for the upgrade regression") + + var messageID int64 + require.NoError(st.DB().QueryRow( + `SELECT id FROM messages WHERE source_message_id = 'html1'`).Scan(&messageID)) + const oldDerived = "quoted parent text\n\nfresh reply" + _, err = st.DB().Exec(st.Rebind(`UPDATE message_bodies SET body_text = ? WHERE message_id = ?`), oldDerived, messageID) + require.NoError(err) + _, err = st.DB().Exec(st.Rebind(`UPDATE messages SET snippet = ? WHERE id = ?`), oldDerived, messageID) + require.NoError(err) + _, err = st.DB().Exec(`DELETE FROM messages_fts WHERE rowid = ?`, messageID) + require.NoError(err) + _, err = st.DB().Exec(`DELETE FROM applied_migrations WHERE name LIKE 'rederive:beeper:signal:%'`) + require.NoError(err) + _, err = st.DB().Exec(`INSERT INTO applied_migrations (name) VALUES ('rederive:beeper:signal:v2')`) + require.NoError(err) + + _, err = imp.Import(context.Background(), ImportOptions{AccountID: "signal"}) + require.NoError(err) + + var body, snippet string + require.NoError(st.DB().QueryRow(`SELECT b.body_text, COALESCE(m.snippet, '') + FROM messages m JOIN message_bodies b ON b.message_id = m.id + WHERE m.id = ?`, messageID).Scan(&body, &snippet)) + assert.Equal("fresh reply", body, "the current repair version must run after v2") + assert.Equal("fresh reply", snippet) + var ftsHits int + require.NoError(st.DB().QueryRow( + `SELECT COUNT(*) FROM messages_fts WHERE rowid = ? AND messages_fts MATCH 'fresh'`, messageID).Scan(&ftsHits)) + assert.Equal(1, ftsHits, "the current repair version must restore FTS after v2") + require.NoError(st.DB().QueryRow( + `SELECT COUNT(*) FROM messages_fts WHERE rowid = ? AND messages_fts MATCH 'quoted'`, messageID).Scan(&ftsHits)) + assert.Zero(ftsHits, "the parent fallback must not remain searchable as reply text") +} + +// beeperSourceID returns the archive's single Beeper source row. +func beeperSourceID(t *testing.T, st *store.Store) int64 { + t.Helper() + var id int64 + require.NoError(t, st.DB().QueryRow( + `SELECT id FROM sources WHERE source_type = 'beeper'`).Scan(&id)) + return id +} diff --git a/internal/beeper/syncstate.go b/internal/beeper/syncstate.go index 9be3cdb59..783a4486d 100644 --- a/internal/beeper/syncstate.go +++ b/internal/beeper/syncstate.go @@ -43,6 +43,10 @@ type SyncState struct { // ListWatermark is the max chat lastActivity observed (RFC3339); the next // incremental run enumerates only chats active after it. ListWatermark string `json:"list_watermark,omitempty"` + // LastTailScan is when this source last re-probed completed chats for + // history Beeper backfilled after they were marked done (RFC3339). See + // tailScanInterval. + LastTailScan string `json:"last_tail_scan,omitempty"` } func NewSyncState() *SyncState { @@ -109,4 +113,7 @@ func (s *SyncState) Merge(other *SyncState) { if other.ListWatermark > s.ListWatermark { s.ListWatermark = other.ListWatermark } + if other.LastTailScan > s.LastTailScan { + s.LastTailScan = other.LastTailScan + } } diff --git a/internal/beeper/types.go b/internal/beeper/types.go index 6c5894e7c..e1743b392 100644 --- a/internal/beeper/types.go +++ b/internal/beeper/types.go @@ -181,6 +181,14 @@ type ImportSummary struct { // new rows. MessagesAdded int64 ReactionsRefreshed int64 + // ChatsReopened counts completed chats whose backfill was resumed because + // Beeper had since added older history behind them (see tailScanInterval). + ChatsReopened int64 + // BodiesRepaired and AttachmentsRetagged report the one-time re-derivation + // of rows written by an older build, run before this sync (see + // rederive.RunIfStale). Both are zero once an archive has caught up. + BodiesRepaired int64 + AttachmentsRetagged int64 // AttachmentsDownloaded counts media stored this run; // AttachmentsPending counts failed/deferred downloads that left a // retry marker (see backfill-beeper-media). diff --git a/internal/query/cache_state.go b/internal/query/cache_state.go index 6ba0f13c0..2c96143d2 100644 --- a/internal/query/cache_state.go +++ b/internal/query/cache_state.go @@ -19,10 +19,10 @@ import ( // relationship activity, people, domain, and daily read model; version 16 // adds has_attachments to the relationship activity dataset; version 17 adds // the envelope address snapshot (email_address) to message_recipients; version -// 18 adds participant-directory revision tracking so a pre-upgrade cache cannot -// be mistaken for one that has observed later participant metadata changes. -// Schema bumps force a full rebuild before readers use an older publication. -const CacheSchemaVersion = 18 +// 18 adds participant-directory revision tracking; version 19 exports +// attachment_metadata for raw attachment queries. Schema bumps force a full +// rebuild before readers use an older publication. +const CacheSchemaVersion = 19 // CacheSyncState is the commit marker written after a complete analytics // cache publication. SQLite remains authoritative; these watermarks only @@ -37,6 +37,11 @@ type CacheSyncState struct { LastFailedSyncRunCount int64 `json:"last_failed_sync_run_count,omitempty"` LastFailedSyncRunIDSum int64 `json:"last_failed_sync_run_id_sum,omitempty"` IdentityRevision int64 `json:"identity_revision,omitempty"` + // DerivedDataRevision tracks offline repairs that rewrite existing + // message, snippet, search, or attachment facts. Those rows are already + // inside the committed message ID boundary, so drift requires a full cache + // rebuild rather than an incremental append. + DerivedDataRevision int64 `json:"derived_data_revision,omitempty"` // AccountIdentityRevision tracks identity mutations that invalidate // baked message data — confirming or removing a "me" address, and // participant merges (which repoint messages.sender_id) — separately @@ -100,7 +105,7 @@ func (e *CacheUnavailableError) Unwrap() error { return ErrCacheUnavailable } // Revision identifies one committed cache publication. It intentionally uses // only commit-marker fields, never ambient filesystem state. func (s CacheSyncState) Revision() string { - payload := fmt.Sprintf("v=%d|message=%d|watermark=%s|run=%d|add=%d|update=%d|fail_count=%d|fail_sum=%d|identity=%d|account_identity=%d|participant_identifier=%d|participant_display_name=%d|published=%s", + payload := fmt.Sprintf("v=%d|message=%d|watermark=%s|run=%d|add=%d|update=%d|fail_count=%d|fail_sum=%d|identity=%d|derived_data=%d|account_identity=%d|participant_identifier=%d|participant_display_name=%d|published=%s", s.SchemaVersion, s.LastMessageID, s.LastSyncAt.UTC().Format(time.RFC3339Nano), @@ -110,6 +115,7 @@ func (s CacheSyncState) Revision() string { s.LastFailedSyncRunCount, s.LastFailedSyncRunIDSum, s.IdentityRevision, + s.DerivedDataRevision, s.AccountIdentityRevision, s.ParticipantIdentifierRevision, s.ParticipantDisplayNameRevision, diff --git a/internal/query/cache_state_test.go b/internal/query/cache_state_test.go index 76f88b49b..3c6010d7e 100644 --- a/internal/query/cache_state_test.go +++ b/internal/query/cache_state_test.go @@ -215,6 +215,9 @@ func TestCacheRevisionUsesOnlyCommittedStateWatermarks(t *testing.T) { changed.IdentityRevision++ assert.NotEqual(revision, changed.Revision()) changed = state + changed.DerivedDataRevision++ + assert.NotEqual(revision, changed.Revision()) + changed = state changed.ParticipantIdentifierRevision++ assert.NotEqual(revision, changed.Revision()) changed = state diff --git a/internal/query/views.go b/internal/query/views.go index ced17e9dc..60baa1f54 100644 --- a/internal/query/views.go +++ b/internal/query/views.go @@ -287,11 +287,18 @@ func createBaseViews(db *sql.DB, analyticsDir string, optCols map[string]map[str "CAST(size AS BIGINT) AS size", "CAST(filename AS VARCHAR) AS filename", }, - optionalCols: []optionalCol{{ - name: "mime_type", - replaceExpr: "COALESCE(CAST(mime_type AS VARCHAR), '') AS mime_type", - defaultExpr: "'' AS mime_type", - }}, + optionalCols: []optionalCol{ + { + name: "mime_type", + replaceExpr: "COALESCE(CAST(mime_type AS VARCHAR), '') AS mime_type", + defaultExpr: "'' AS mime_type", + }, + { + name: "attachment_metadata", + replaceExpr: "TRY_CAST(attachment_metadata AS VARCHAR) AS attachment_metadata", + defaultExpr: "NULL::VARCHAR AS attachment_metadata", + }, + }, }, probe: colsFor("attachments"), }, diff --git a/internal/query/views_test.go b/internal/query/views_test.go index 131c5d7a1..e5f420bee 100644 --- a/internal/query/views_test.go +++ b/internal/query/views_test.go @@ -62,6 +62,7 @@ func TestRegisterViews_BaseViews(t *testing.T) { }) builder.AddFrom(msgID, partID, "Bob") builder.AddMessageLabel(msgID, lblID) + builder.AddAttachment(msgID, 500, "preview.jpg") dir, cleanup := builder.Build() defer cleanup() @@ -94,6 +95,14 @@ func TestRegisterViews_BaseViews(t *testing.T) { ).Scan(&id, &subject, &attachmentCount, &messageType) require.NoError(err, "scan messages") assert.Equal("Hello", subject) + + var attachmentMetadata sql.NullString + err = engine.db.QueryRowContext( + context.Background(), + "SELECT attachment_metadata FROM attachments LIMIT 1", + ).Scan(&attachmentMetadata) + require.NoError(err, "legacy attachment cache must expose attachment_metadata") + assert.False(attachmentMetadata.Valid, "legacy attachment rows default to unclassified") } func TestRegisterViews_ConvenienceViews(t *testing.T) { diff --git a/internal/rederive/rederive.go b/internal/rederive/rederive.go new file mode 100644 index 000000000..b570b625b --- /dev/null +++ b/internal/rederive/rederive.go @@ -0,0 +1,153 @@ +// Package rederive re-computes stored message columns from the verbatim +// provider payloads archived alongside them. +// +// Importers derive columns — body text, snippets, the search index, attachment +// classification — from a provider's payload at import time, then archive that +// payload as-is (message_raw.raw_format). Improving how a column is derived +// therefore leaves every already-archived message stale, and re-syncing repairs +// it only where the provider still holds the message and only at network speed. +// +// A registered pass reads the archive instead: offline, bounded by disk rather +// than an API, and able to repair messages the provider has since dropped. +// Registration lives here rather than in each importer so one command and one +// upgrade path serve all of them. +package rederive + +import ( + "context" + "errors" + "fmt" + "sort" + "time" + + "go.kenn.io/msgvault/internal/store" +) + +// Summary reports what a re-derivation pass changed. +type Summary struct { + Duration time.Duration + MessagesScanned int64 + BodiesRewritten int64 + AttachmentsTagged int64 + // Undecodable counts archived payloads that could not be parsed; they are + // left untouched. + Undecodable int64 + Errors int64 +} + +// Add accumulates other into s, except Duration, which the caller owns. +func (s *Summary) Add(other *Summary) { + if other == nil { + return + } + s.MessagesScanned += other.MessagesScanned + s.BodiesRewritten += other.BodiesRewritten + s.AttachmentsTagged += other.AttachmentsTagged + s.Undecodable += other.Undecodable + s.Errors += other.Errors +} + +// Func re-derives every archived message of one source. progress may be nil. +type Func func(ctx context.Context, s *store.Store, sourceID int64, progress func(string)) (*Summary, error) + +type entry struct { + fn Func + version string +} + +var registry = map[string]entry{} + +// Register associates a source type with its re-derivation pass. +// +// version identifies the derivation logic, not the schema: bump it whenever a +// change would produce different output for the same payload, so archives heal +// on their next sync. Registering a source type twice is a programming error +// and panics, since the second pass would silently shadow the first. +func Register(sourceType, version string, fn Func) { + if _, dup := registry[sourceType]; dup { + panic(fmt.Sprintf("rederive: source type %q registered twice", sourceType)) + } + registry[sourceType] = entry{fn: fn, version: version} +} + +// Lookup returns the pass registered for a source type. +func Lookup(sourceType string) (Func, string, bool) { + e, ok := registry[sourceType] + return e.fn, e.version, ok +} + +// SourceTypes lists every registered source type, sorted for stable output. +func SourceTypes() []string { + out := make([]string, 0, len(registry)) + for k := range registry { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// LedgerKey names the applied_migrations entry recording that a source has +// been re-derived at a given version. Keying per source (rather than per source +// type) lets one account heal without blocking or repeating the others. +func LedgerKey(sourceType, identifier, version string) string { + return fmt.Sprintf("rederive:%s:%s:%s", sourceType, identifier, version) +} + +// Run executes a source's pass and records it in the ledger, whether or not the +// ledger already held it. This is the on-demand path; recording still matters +// here, or the next sync would repeat a full scan of an archive that is already +// current. +// +// The pass is recorded only on success, so a failed attempt is retried later +// rather than being silently marked done. +func Run( + ctx context.Context, s *store.Store, sourceType, identifier string, sourceID int64, progress func(string), +) (*Summary, error) { + fn, version, ok := Lookup(sourceType) + if !ok { + return nil, fmt.Errorf("no re-derivation pass registered for source type %q", sourceType) + } + sum, err := fn(ctx, s, sourceID, progress) + if err != nil { + // A pass can fail after earlier message transactions committed. Make + // those partial authoritative writes visible to cache maintenance even + // though the repair remains retryable. + if sum != nil && (sum.MessagesScanned > 0 || sum.Errors > 0) { + return sum, errors.Join(err, s.AdvanceDerivedDataRevision()) + } + return sum, err + } + if sum != nil && sum.Errors > 0 { + return sum, s.AdvanceDerivedDataRevision() + } + ledgerKey := LedgerKey(sourceType, identifier, version) + if sum == nil || sum.MessagesScanned == 0 { + // New and empty sources have no existing derived rows to invalidate. + // Record the pass so sync does not repeat it, but keep a current cache + // valid after the source's first import. + return sum, s.MarkMigrationApplied(ledgerKey) + } + if err := s.MarkMigrationAppliedWithDerivedDataRevision(ledgerKey); err != nil { + return sum, err + } + return sum, nil +} + +// RunIfStale runs the registered pass for a source unless the ledger already +// records it at the current version. ran reports whether the pass actually +// executed: a source type with no registered pass, or one already recorded, is +// a no-op. This is the upgrade path, called from a sync. +func RunIfStale( + ctx context.Context, s *store.Store, sourceType, identifier string, sourceID int64, progress func(string), +) (sum *Summary, ran bool, err error) { + _, version, ok := Lookup(sourceType) + if !ok { + return nil, false, nil + } + applied, err := s.IsMigrationApplied(LedgerKey(sourceType, identifier, version)) + if err != nil || applied { + return nil, false, err + } + sum, err = Run(ctx, s, sourceType, identifier, sourceID, progress) + return sum, true, err +} diff --git a/internal/rederive/rederive_test.go b/internal/rederive/rederive_test.go new file mode 100644 index 000000000..0525146c1 --- /dev/null +++ b/internal/rederive/rederive_test.go @@ -0,0 +1,206 @@ +package rederive + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/msgvault/internal/store" + "go.kenn.io/msgvault/internal/testutil" +) + +func TestLedgerKeyIsPerSourceAndVersion(t *testing.T) { + assert := assert.New(t) + assert.Equal("rederive:beeper:instagramgo:v1", LedgerKey("beeper", "instagramgo", "v1")) + // One account healing must not mark another as done, and a version bump + // must make an already-healed source stale again. + assert.NotEqual(LedgerKey("beeper", "instagramgo", "v1"), LedgerKey("beeper", "whatsapp", "v1")) + assert.NotEqual(LedgerKey("beeper", "instagramgo", "v1"), LedgerKey("beeper", "instagramgo", "v2")) +} + +func TestRegisterRejectsDuplicates(t *testing.T) { + assert := assert.New(t) + noop := func(context.Context, *store.Store, int64, func(string)) (*Summary, error) { return &Summary{}, nil } + Register("test-dup", "v1", noop) + t.Cleanup(func() { delete(registry, "test-dup") }) + // A silently shadowed pass would leave archives unrepaired with no signal. + assert.Panics(func() { Register("test-dup", "v1", noop) }) +} + +func TestLookupReportsUnregisteredTypes(t *testing.T) { + _, _, ok := Lookup("not-a-source-type") + assert.False(t, ok) +} + +func TestSummaryAddAccumulates(t *testing.T) { + assert := assert.New(t) + total := &Summary{} + total.Add(&Summary{MessagesScanned: 3, BodiesRewritten: 2, Errors: 1}) + total.Add(&Summary{MessagesScanned: 4, AttachmentsTagged: 5, Undecodable: 1}) + total.Add(nil) + assert.Equal(int64(7), total.MessagesScanned) + assert.Equal(int64(2), total.BodiesRewritten) + assert.Equal(int64(5), total.AttachmentsTagged) + assert.Equal(int64(1), total.Undecodable) + assert.Equal(int64(1), total.Errors) +} + +// TestRunIfStaleRecordsAndSkips covers the gate that keeps a healed archive +// from re-deriving on every sync, and the retry left behind by a failed pass. +func TestRunIfStaleRecordsAndSkips(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + st := testutil.NewTestStore(t) + calls := 0 + Register("test-gate", "v1", func(context.Context, *store.Store, int64, func(string)) (*Summary, error) { + calls++ + return &Summary{MessagesScanned: 10}, nil + }) + t.Cleanup(func() { delete(registry, "test-gate") }) + + sum, ran, err := RunIfStale(context.Background(), st, "test-gate", "acct", 1, nil) + require.NoError(err) + require.True(ran) + require.NotNil(sum) + assert.Equal(int64(10), sum.MessagesScanned) + assert.Equal(1, calls) + assert.Equal(int64(1), derivedDataRevision(t, st), + "a sync-triggered repair must invalidate the analytics cache") + + _, ran, err = RunIfStale(context.Background(), st, "test-gate", "acct", 1, nil) + require.NoError(err) + assert.False(ran, "an already-recorded pass must not run again") + assert.Equal(1, calls) + assert.Equal(int64(1), derivedDataRevision(t, st), + "skipping an applied pass must not invalidate the cache again") + + // A different account of the same type is tracked separately. + _, ran, err = RunIfStale(context.Background(), st, "test-gate", "other", 2, nil) + require.NoError(err) + assert.True(ran) + assert.Equal(2, calls) + assert.Equal(int64(2), derivedDataRevision(t, st), + "each completed source repair must advance cache freshness") +} + +func TestRunIfStaleEmptyPassRecordsWithoutAdvancingRevision(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + st := testutil.NewTestStore(t) + calls := 0 + Register("test-empty", "v1", func(context.Context, *store.Store, int64, func(string)) (*Summary, error) { + calls++ + return &Summary{}, nil + }) + t.Cleanup(func() { delete(registry, "test-empty") }) + + sum, ran, err := RunIfStale(context.Background(), st, "test-empty", "acct", 1, nil) + require.NoError(err) + require.True(ran) + require.NotNil(sum) + assert.Zero(sum.MessagesScanned) + assert.Equal(1, calls) + assert.Zero(derivedDataRevision(t, st), + "an empty source has no derived rows that could stale the analytics cache") + + applied, err := st.IsMigrationApplied(LedgerKey("test-empty", "acct", "v1")) + require.NoError(err) + assert.True(applied, "an empty successful pass must still be recorded") + + _, ran, err = RunIfStale(context.Background(), st, "test-empty", "acct", 1, nil) + require.NoError(err) + assert.False(ran, "the recorded empty pass must not repeat on every sync") + assert.Equal(1, calls) +} + +func TestRunIfStaleRetriesSummaryErrorsWithoutRecordingLedger(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + st := testutil.NewTestStore(t) + calls := 0 + Register("test-partial", "v1", func(context.Context, *store.Store, int64, func(string)) (*Summary, error) { + calls++ + return &Summary{MessagesScanned: 3, Errors: 1}, nil + }) + t.Cleanup(func() { delete(registry, "test-partial") }) + + sum, ran, err := RunIfStale(context.Background(), st, "test-partial", "acct", 1, nil) + require.NoError(err) + require.True(ran) + require.NotNil(sum) + assert.Equal(int64(1), sum.Errors) + + applied, err := st.IsMigrationApplied(LedgerKey("test-partial", "acct", "v1")) + require.NoError(err) + assert.False(applied, "a pass with row-level errors must remain retryable") + assert.Equal(int64(1), derivedDataRevision(t, st), + "partial writes must still invalidate the analytics cache") + + _, ran, err = RunIfStale(context.Background(), st, "test-partial", "acct", 1, nil) + require.NoError(err) + assert.True(ran, "the next sync must retry an incomplete pass") + assert.Equal(2, calls) + assert.Equal(int64(2), derivedDataRevision(t, st)) +} + +// TestRunRecordsSoASyncDoesNotRepeatIt covers the on-demand path: it runs even +// when the ledger already holds the pass, but still records it, so repairing by +// hand does not leave the next sync re-scanning a current archive. +func TestRunRecordsSoASyncDoesNotRepeatIt(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + + st := testutil.NewTestStore(t) + calls := 0 + Register("test-run", "v1", func(context.Context, *store.Store, int64, func(string)) (*Summary, error) { + calls++ + return &Summary{MessagesScanned: 7}, nil + }) + t.Cleanup(func() { delete(registry, "test-run") }) + + sum, err := Run(context.Background(), st, "test-run", "acct", 1, nil) + require.NoError(err) + assert.Equal(int64(7), sum.MessagesScanned) + assert.Equal(1, calls) + assert.Equal(int64(1), derivedDataRevision(t, st), + "an on-demand repair must invalidate the analytics cache") + + // A later sync sees the archive as current. + _, ran, err := RunIfStale(context.Background(), st, "test-run", "acct", 1, nil) + require.NoError(err) + assert.False(ran, "an on-demand repair must record the ledger") + assert.Equal(1, calls) + assert.Equal(int64(1), derivedDataRevision(t, st)) + + // Asking again by hand still re-runs: that is the point of the escape hatch. + _, err = Run(context.Background(), st, "test-run", "acct", 1, nil) + require.NoError(err) + assert.Equal(2, calls) + assert.Equal(int64(2), derivedDataRevision(t, st)) +} + +func derivedDataRevision(t *testing.T, st *store.Store) int64 { + t.Helper() + var revision int64 + require.NoError(t, st.DB().QueryRow(` + SELECT COALESCE(CAST(( + SELECT value FROM archive_metadata WHERE key = 'derived_data_revision' + ) AS BIGINT), 0) + `).Scan(&revision)) + return revision +} + +func TestRunRejectsUnregisteredType(t *testing.T) { + _, err := Run(context.Background(), testutil.NewTestStore(t), "no-such-type", "acct", 1, nil) + require.Error(t, err) +} + +func TestRunIfStaleIsNoOpForUnregisteredType(t *testing.T) { + _, ran, err := RunIfStale(context.Background(), testutil.NewTestStore(t), "no-such-type", "acct", 1, nil) + require.NoError(t, err) + assert.False(t, ran) +} diff --git a/internal/store/attachments.go b/internal/store/attachments.go index 0252828a6..2fb9293c7 100644 --- a/internal/store/attachments.go +++ b/internal/store/attachments.go @@ -32,7 +32,8 @@ func (s *Store) replaceMessageProviderAttachments(messageID int64, providerPrefi func (s *Store) messageProviderAttachments(messageID int64, providerPrefix string) (map[string]AttachmentRef, error) { rows, err := s.db.Query(` SELECT COALESCE(filename, ''), COALESCE(mime_type, ''), storage_path, COALESCE(content_hash, ''), size, source_attachment_id, - COALESCE(media_type, ''), COALESCE(width, 0), COALESCE(height, 0), COALESCE(duration_ms, 0) + COALESCE(media_type, ''), COALESCE(width, 0), COALESCE(height, 0), COALESCE(duration_ms, 0), + COALESCE(CAST(attachment_metadata AS TEXT), '') FROM attachments WHERE message_id = ? AND source_attachment_id LIKE ? `, messageID, providerPrefix+"%") @@ -48,7 +49,7 @@ func (s *Store) messageProviderAttachments(messageID int64, providerPrefix strin if err := rows.Scan( &ref.Filename, &ref.MimeType, &ref.StoragePath, &ref.ContentHash, &size, &ref.SourceAttachmentID, &ref.MediaType, &ref.Width, - &ref.Height, &ref.DurationMS, + &ref.Height, &ref.DurationMS, &ref.Metadata, ); err != nil { return nil, err } diff --git a/internal/store/attachments_test.go b/internal/store/attachments_test.go index 17abb945d..ebe7b326a 100644 --- a/internal/store/attachments_test.go +++ b/internal/store/attachments_test.go @@ -351,6 +351,7 @@ func TestReplaceAndListMessageDiscordAttachments(t *testing.T) { MediaType: "image", Width: 640, Height: 480, + Metadata: `{"source_url":"https://example.com/post/1"}`, }, "discord:attachment-2": { Filename: "later.bin", @@ -367,6 +368,10 @@ func TestReplaceAndListMessageDiscordAttachments(t *testing.T) { got, err := st.MessageDiscordAttachments(messageID) require.NoError(err) + assert.JSONEq(want["discord:attachment-1"].Metadata, got["discord:attachment-1"].Metadata) + wantMetadata := want["discord:attachment-1"] + wantMetadata.Metadata = got["discord:attachment-1"].Metadata + want["discord:attachment-1"] = wantMetadata assert.Equal(want, got) keep := want["discord:attachment-2"] diff --git a/internal/store/derived_data_revision.go b/internal/store/derived_data_revision.go new file mode 100644 index 000000000..7c9b966d4 --- /dev/null +++ b/internal/store/derived_data_revision.go @@ -0,0 +1,77 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "strconv" +) + +const derivedDataRevisionKey = "derived_data_revision" + +// DerivedDataRevision returns the revision of existing message facts changed +// by offline re-derivation. Analytics caches stamp this value when they export +// message and attachment Parquet; a mismatch requires a full rebuild because +// incremental publication cannot rewrite already-exported rows. +func (s *Store) DerivedDataRevision() (int64, error) { + var value string + err := s.db.QueryRow( + `SELECT value FROM archive_metadata WHERE key = ?`, derivedDataRevisionKey, + ).Scan(&value) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("read derived-data revision: %w", err) + } + revision, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, fmt.Errorf("parse derived-data revision %q: %w", value, err) + } + return revision, nil +} + +func (s *Store) bumpDerivedDataRevision(tx *loggedTx) error { + if _, err := tx.Exec(s.dialect.InsertOrIgnore( + `INSERT OR IGNORE INTO archive_metadata (key, value) VALUES (?, '0')`), + derivedDataRevisionKey); err != nil { + return fmt.Errorf("seed derived-data revision: %w", err) + } + if _, err := tx.Exec(` + UPDATE archive_metadata + SET value = CAST(CAST(value AS INTEGER) + 1 AS TEXT) + WHERE key = ? + `, derivedDataRevisionKey); err != nil { + return fmt.Errorf("bump derived-data revision: %w", err) + } + return nil +} + +// AdvanceDerivedDataRevision records a repair attempt that may have committed +// changes but was not complete enough to enter the migration ledger. The next +// cache maintenance pass must still publish those partial, authoritative rows. +func (s *Store) AdvanceDerivedDataRevision() error { + return s.withTx(func(tx *loggedTx) error { + return s.bumpDerivedDataRevision(tx) + }) +} + +// MarkMigrationAppliedWithDerivedDataRevision atomically records a completed +// re-derivation and advances the cache-visible revision. The ledger can never +// claim a repair is complete without also making an older analytics cache +// stale. +func (s *Store) MarkMigrationAppliedWithDerivedDataRevision(name string) error { + return s.withTx(func(tx *loggedTx) error { + if err := s.bumpDerivedDataRevision(tx); err != nil { + return err + } + _, err := tx.Exec( + s.dialect.InsertOrIgnore(`INSERT OR IGNORE INTO applied_migrations (name) VALUES (?)`), + name, + ) + if err != nil { + return fmt.Errorf("mark migration %q applied: %w", name, err) + } + return nil + }) +} diff --git a/internal/store/dialect.go b/internal/store/dialect.go index 5a5bb57cc..db030c360 100644 --- a/internal/store/dialect.go +++ b/internal/store/dialect.go @@ -372,6 +372,12 @@ type Dialect interface { // "column is of type jsonb but expression is of type text". JSONBindExpr() string + // JSONIsDistinctExpr returns a null-safe comparison between a JSON column + // and one bound JSON value. PostgreSQL compares parsed JSONB values rather + // than their differently formatted text renderings; SQLite compares its + // stored JSON text directly. + JSONIsDistinctExpr(col string) string + // BeginExclusive opens a transaction on conn that blocks concurrent // writers to the tables sync code touches (sync_runs in particular, // so StartSync's INSERT cannot run until COMMIT/ROLLBACK). Readers diff --git a/internal/store/dialect_pg.go b/internal/store/dialect_pg.go index 180094a4e..3721918d2 100644 --- a/internal/store/dialect_pg.go +++ b/internal/store/dialect_pg.go @@ -287,6 +287,10 @@ func (d *PostgreSQLDialect) BoolTrueExpr(col string) string { return col } // mismatch on the sources.sync_config write path. func (d *PostgreSQLDialect) JSONBindExpr() string { return "?::JSONB" } +func (d *PostgreSQLDialect) JSONIsDistinctExpr(col string) string { + return col + " IS DISTINCT FROM ?::JSONB" +} + // BuildFTSArg formats search terms for to_tsquery: each term is split // into letter/digit-only lexemes via sqldialect.EscapeTSQueryTerm so // punctuation like `-`, `.`, `@` (which would otherwise produce diff --git a/internal/store/dialect_sqlite.go b/internal/store/dialect_sqlite.go index 0db176c53..466991bdc 100644 --- a/internal/store/dialect_sqlite.go +++ b/internal/store/dialect_sqlite.go @@ -304,6 +304,8 @@ func (d *SQLiteDialect) BoolTrueExpr(col string) string { return col + " = 1" } // JSONBindExpr is "?" on SQLite — JSON columns are plain TEXT. func (d *SQLiteDialect) JSONBindExpr() string { return "?" } +func (d *SQLiteDialect) JSONIsDistinctExpr(col string) string { return col + " IS NOT ?" } + // BuildFTSArg formats search terms as an FTS5 MATCH argument: each // term double-quote-escaped, suffixed with "*" for prefix match, and // space-joined (FTS5 treats space as implicit AND). Embedded "*" is diff --git a/internal/store/messages.go b/internal/store/messages.go index f971cf5f4..92443defe 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -3846,6 +3846,10 @@ type AttachmentRef struct { Width int64 Height int64 DurationMS int64 + // Metadata is importer-supplied JSON stored in attachments.attachment_metadata + // (e.g. the source URL a link-preview attachment was forwarded from). Empty + // stores NULL; callers own the shape and must supply valid JSON. + Metadata string } // replaceMessageAttachmentsWhere atomically deletes a message's attachment @@ -3865,11 +3869,12 @@ func (s *Store) replaceMessageAttachmentsWhere( } if _, err := tx.Exec(fmt.Sprintf(` INSERT INTO attachments (message_id, filename, mime_type, storage_path, content_hash, size, source_attachment_id, - media_type, width, height, duration_ms, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, %s) + media_type, width, height, duration_ms, attachment_metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, %s, %s) ON CONFLICT (message_id, content_hash) WHERE content_hash IS NOT NULL AND content_hash != '' DO NOTHING - `, s.dialect.Now()), messageID, ref.Filename, ref.MimeType, ref.StoragePath, ref.ContentHash, int64(ref.Size), ref.SourceAttachmentID, - nullIfEmpty(ref.MediaType), nullIfZero(ref.Width), nullIfZero(ref.Height), nullIfZero(ref.DurationMS)); err != nil { + `, s.dialect.JSONBindExpr(), s.dialect.Now()), messageID, ref.Filename, ref.MimeType, ref.StoragePath, ref.ContentHash, int64(ref.Size), ref.SourceAttachmentID, + nullIfEmpty(ref.MediaType), nullIfZero(ref.Width), nullIfZero(ref.Height), nullIfZero(ref.DurationMS), + nullIfEmpty(ref.Metadata)); err != nil { return err } } @@ -3920,6 +3925,137 @@ func (s *Store) MessageBeeperAttachments(messageID int64) (map[string]Attachment return s.messageProviderAttachments(messageID, "beeper:") } +// ArchivedRawMessage is one archived message paired with the verbatim provider +// payload stored for it, decompressed. +type ArchivedRawMessage struct { + MessageID int64 + ConversationID int64 + RawData []byte + // BodyText is the currently stored plain-text body, so a caller + // re-deriving it can skip rows that would not change. + BodyText string +} + +// ScanArchivedRawMessages returns up to limit archived messages for a source +// whose raw payload is in the given format, ordered by message ID and starting +// after afterID. Paging by ID keeps a full-archive walk bounded in memory; +// callers loop until an empty batch comes back. +func (s *Store) ScanArchivedRawMessages(sourceID int64, format string, afterID int64, limit int) ([]ArchivedRawMessage, error) { + rows, err := s.db.Query(s.Rebind(` + SELECT m.id, m.conversation_id, r.raw_data, r.compression, COALESCE(b.body_text, '') + FROM messages m + JOIN message_raw r ON r.message_id = m.id + LEFT JOIN message_bodies b ON b.message_id = m.id + WHERE m.source_id = ? AND r.raw_format = ? AND m.id > ? + ORDER BY m.id + LIMIT ? + `), sourceID, format, afterID, limit) + if err != nil { + return nil, fmt.Errorf("scan archived raw messages: %w", err) + } + defer func() { _ = rows.Close() }() + var out []ArchivedRawMessage + for rows.Next() { + var item ArchivedRawMessage + var raw []byte + var compression sql.NullString + if err := rows.Scan(&item.MessageID, &item.ConversationID, &raw, &compression, &item.BodyText); err != nil { + return nil, err + } + if compression.Valid && compression.String == "zlib" { + r, zerr := zlib.NewReader(bytes.NewReader(raw)) + if zerr != nil { + return nil, fmt.Errorf("zlib reader for message %d: %w", item.MessageID, zerr) + } + raw, err = io.ReadAll(r) + _ = r.Close() + if err != nil { + return nil, fmt.Errorf("decompress message %d: %w", item.MessageID, err) + } + } + item.RawData = raw + out = append(out, item) + } + return out, rows.Err() +} + +// SetBeeperAttachmentMetadata replaces attachment_metadata on a message's +// Beeper-managed attachment rows, returning how many rows changed. Empty +// metadata clears the column. Rows already holding the value are left alone so +// a repeated call reports no change; updating in place (rather than replacing +// rows) leaves the stored blobs untouched. +func (s *Store) SetBeeperAttachmentMetadata(messageID int64, metadata string) (int64, error) { + res, err := s.db.Exec(s.Rebind(fmt.Sprintf(` + UPDATE attachments SET attachment_metadata = %s + WHERE message_id = ? AND source_attachment_id LIKE 'beeper:%%' + AND %s + `, s.dialect.JSONBindExpr(), s.dialect.JSONIsDistinctExpr("attachment_metadata"))), + nullIfEmpty(metadata), messageID, nullIfEmpty(metadata)) + if err != nil { + return 0, fmt.Errorf("set beeper attachment metadata: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("set beeper attachment metadata: rows affected: %w", err) + } + return n, nil +} + +// UpdateMessageDerivedText atomically updates the text fields derived from one +// provider payload. A body, snippet, or FTS failure rolls the whole update +// back, so callers can safely retry every derived field together. +func (s *Store) UpdateMessageDerivedText( + messageID int64, bodyText, bodyHTML, snippet sql.NullString, fts FTSDoc, +) error { + fts.MessageID = messageID + return s.withTx(func(tx *loggedTx) error { + if err := upsertMessageBody(tx, s.dialect, s.fts5Available, messageID, bodyText, bodyHTML); err != nil { + return fmt.Errorf("update derived message body: %w", err) + } + if _, err := tx.Exec(`UPDATE messages SET snippet = ? WHERE id = ?`, snippet, messageID); err != nil { + return fmt.Errorf("update derived message snippet: %w", err) + } + if s.fts5Available { + if err := s.dialect.FTSUpsert(tx, fts); err != nil { + return fmt.Errorf("update derived message FTS: %w", err) + } + } + return nil + }) +} + +// ArchivedSourceMessageIDs returns the subset of sourceMessageIDs already +// archived for a source. Used to decide whether a page fetched from the +// provider contains anything new without re-persisting it first. +func (s *Store) ArchivedSourceMessageIDs(sourceID int64, sourceMessageIDs []string) (map[string]struct{}, error) { + out := make(map[string]struct{}) + if len(sourceMessageIDs) == 0 { + return out, nil + } + placeholders := make([]string, len(sourceMessageIDs)) + args := make([]any, 0, len(sourceMessageIDs)+1) + args = append(args, sourceID) + for i, id := range sourceMessageIDs { + placeholders[i] = "?" + args = append(args, id) + } + query := s.Rebind(`SELECT source_message_id FROM messages WHERE source_id = ? AND source_message_id IN (` + + strings.Join(placeholders, ",") + `)`) + rows, err := s.db.Query(query, args...) + if err != nil { + return nil, fmt.Errorf("look up archived source message IDs: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + out[id] = struct{}{} + } + return out, rows.Err() +} + // SourceMessageRef locates an archived message at its source: the source // message ID, its conversation's source ID, and the archived timestamp. type SourceMessageRef struct { diff --git a/internal/textutil/encoding.go b/internal/textutil/encoding.go index fb0a9033a..e794271e6 100644 --- a/internal/textutil/encoding.go +++ b/internal/textutil/encoding.go @@ -158,6 +158,18 @@ func FirstLine(s string) string { // the raw leading byte, so that UTF-8 encoded C1 chars (e.g., U+009B CSI // encoded as 0xC2 0x9B) are correctly stripped. func SanitizeTerminal(s string) string { + return sanitizeTerminal(s, false) +} + +// SanitizeTerminalMultiline strips ANSI escape sequences and C0/C1 control +// characters while preserving line feeds. Carriage returns are discarded so +// untrusted text cannot overwrite terminal output, and CRLF is normalized to +// LF. Use SanitizeTerminal for single-line terminal output. +func SanitizeTerminalMultiline(s string) string { + return sanitizeTerminal(s, true) +} + +func sanitizeTerminal(s string, multiline bool) string { var b strings.Builder b.Grow(len(s)) i := 0 @@ -205,17 +217,27 @@ func SanitizeTerminal(s string) string { continue } - // Allow tab; strip newline and carriage return (all callers use this - // in single-line contexts such as TUI rows and progress output where - // \r can overwrite lines and \n can break layout). + // Allow tab. Multiline output keeps LF but always discards CR so an + // untrusted value cannot return the cursor to the start of a line. if r == '\t' { b.WriteRune(r) i += size continue } - if r == '\n' || r == '\r' { - // Replace with space to preserve word boundaries. - b.WriteByte(' ') + if r == '\n' { + if multiline { + b.WriteByte('\n') + } else { + b.WriteByte(' ') + } + i += size + continue + } + if r == '\r' { + if !multiline { + // Replace with space to preserve word boundaries. + b.WriteByte(' ') + } i += size continue } diff --git a/internal/textutil/encoding_test.go b/internal/textutil/encoding_test.go index 064d42b8a..673f02a10 100644 --- a/internal/textutil/encoding_test.go +++ b/internal/textutil/encoding_test.go @@ -494,3 +494,24 @@ func TestSanitizeTerminal(t *testing.T) { }) } } + +func TestSanitizeTerminalMultiline(t *testing.T) { + tests := []struct { + name string + input string + want string + }{ + {"preserves newlines", "line1\nline2", "line1\nline2"}, + {"normalizes CRLF", "line1\r\nline2", "line1\nline2"}, + {"strips carriage returns", "over\rwrite", "overwrite"}, + {"strips OSC while preserving surrounding lines", "before\n\x1b]52;c;evil\x07after", "before\nafter"}, + {"strips CSI and C1 controls", "\x1b[31mred\x1b[0m\nleft\u009bright", "red\nleftright"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := SanitizeTerminalMultiline(tt.input) + assert.Equalf(t, tt.want, got, "SanitizeTerminalMultiline(%q)", tt.input) + }) + } +} diff --git a/internal/tui/view.go b/internal/tui/view.go index 3290dd420..8fdcac7aa 100644 --- a/internal/tui/view.go +++ b/internal/tui/view.go @@ -729,9 +729,7 @@ func (m Model) buildDetailLines() []string { if body == "" { body = "(No text content)" } - // Strip carriage returns (CRLF -> LF) to prevent display issues - body = strings.ReplaceAll(body, "\r\n", "\n") - body = strings.ReplaceAll(body, "\r", "") + body = textutil.SanitizeTerminalMultiline(body) bodyLines := wrapText(body, m.width-2) lines = append(lines, bodyLines...) diff --git a/internal/tui/view_render_test.go b/internal/tui/view_render_test.go index 88fdab187..140fc5b96 100644 --- a/internal/tui/view_render_test.go +++ b/internal/tui/view_render_test.go @@ -799,6 +799,20 @@ func TestLayoutFitsTerminalHeight(t *testing.T) { } } +func TestBuildDetailLinesSanitizesMultilineBody(t *testing.T) { + assert := assert.New(t) + model := NewBuilder().WithSize(80, 24).Build() + model.messageDetail = &query.MessageDetail{ + BodyText: "first line\n\x1b]52;c;evil\x07second line\u009b", + } + + output := strings.Join(model.buildDetailLines(), "\n") + assert.Contains(output, "first line\nsecond line") + assert.NotContains(output, "\x1b") + assert.NotContains(output, "\x07") + assert.NotContains(output, "\u009b") +} + // TestScrollClampingAfterResize verifies detailScroll is clamped when max changes. // TestModalCompositingPreservesANSI verifies that modal overlay doesn't corrupt ANSI sequences.