fix(beeper): store plain text, resume backfilled history, classify link previews

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,20 +994,26 @@ 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
 			id AS attachment_id,
 			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",

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")

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()

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)

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