feat: add WhatsApp import from decrypted backup (#136)

Add `import --type whatsapp` command for importing messages from a
decrypted WhatsApp msgstore.db into the msgvault unified schema.

New package internal/whatsapp/:
- Reads msgstore.db as read-only SQLite
- Maps WhatsApp schema to msgvault tables (conversations, participants,
  messages, attachments, reactions, reply threading)
- Batch processing (1000 msgs/batch) with checkpoint/resume
- Optional --contacts for vCard name resolution (update-only, no creation)
- Optional --media-dir for content-addressed media storage
- Imports: text, images, video, audio, voice notes, documents,
  stickers, GIFs, reactions, replies, group participants
- Skips: system messages, calls, location shares, contacts, polls

Security:
- Media path traversal defense (sanitize, reject absolute/.. paths,
  boundary check against mediaDir)
- Streaming hash + copy for media (no io.ReadAll, 100MB max)
- E.164 phone validation (reject non-numeric JIDs, 4-15 digit range)
- File permissions 0600/0750 for attachments
- Per-chat reply map scoping to bound memory

Query engine updates:
- Sender filters in DuckDB and SQLite check both message_recipients
  (email path) and direct sender_id (WhatsApp/chat path)
- Phone number included in sender predicates for from:+447... queries
- MatchesEmpty filters account for sender_id to avoid false positives
- MCP handler routes non-email from values to display name matching
- Parquet cache extended with sender_id, message_type, attachment_count,
  phone_number, title columns
- Cache schema versioning to force rebuild on column layout changes

Store additions:
- EnsureConversationWithType (parameterized conversation_type)
- EnsureParticipantByPhone (E.164 validated, with identifier row)
- UpdateParticipantDisplayNameByPhone (update-only for contacts)
- EnsureConversationParticipant, UpsertReaction, UpsertMessageRawWithFormat

Tested with 1.19M messages (13.5k conversations, April 2016-present).

Author: eddowding

cmd/msgvault/cmd/build_cache.go

@@ -27,10 +27,17 @@ var fullRebuild bool
 // files (_last_sync.json, parquet directories) can corrupt the cache.
 var buildCacheMu sync.Mutex
 
+// cacheSchemaVersion tracks the Parquet schema layout. Bump this whenever
+// columns are added/removed/renamed in the COPY queries below so that
+// incremental builds automatically trigger a full rebuild instead of
+// producing Parquet files with mismatched schemas.
+const cacheSchemaVersion = 2 // v2: added sender_id, message_type, attachment_count, phone_number, title
+
 // syncState tracks the last exported message ID for incremental updates.
 type syncState struct {
 	LastMessageID int64     `json:"last_message_id"`
 	LastSyncAt    time.Time `json:"last_sync_at"`
+	SchemaVersion int       `json:"schema_version,omitempty"`
 }
 
 var buildCacheCmd = &cobra.Command{
@@ -101,8 +108,16 @@ func buildCache(dbPath, analyticsDir string, fullRebuild bool) (*buildResult, er
 		if data, err := os.ReadFile(stateFile); err == nil {
 			var state syncState
 			if json.Unmarshal(data, &state) == nil {
-				lastMessageID = state.LastMessageID
-				fmt.Printf("Incremental export from message_id > %d\n", lastMessageID)
+				if state.SchemaVersion != cacheSchemaVersion {
+					// Schema has changed — force a full rebuild.
+					fmt.Printf("Cache schema version mismatch (have v%d, need v%d). Forcing full rebuild.\n",
+						state.SchemaVersion, cacheSchemaVersion)
+					fullRebuild = true
+					lastMessageID = 0
+				} else {
+					lastMessageID = state.LastMessageID
+					fmt.Printf("Incremental export from message_id > %d\n", lastMessageID)
+				}
 			}
 		}
 	}
@@ -231,7 +246,10 @@ func buildCache(dbPath, analyticsDir string, fullRebuild bool) (*buildResult, er
 			m.sent_at,
 			m.size_estimate,
 			m.has_attachments,
+			COALESCE(TRY_CAST(m.attachment_count AS INTEGER), 0) as attachment_count,
 			m.deleted_from_source_at,
+			m.sender_id,
+			COALESCE(TRY_CAST(m.message_type AS VARCHAR), '') as message_type,
 			CAST(EXTRACT(YEAR FROM m.sent_at) AS INTEGER) as year,
 			CAST(EXTRACT(MONTH FROM m.sent_at) AS INTEGER) as month
 		FROM sqlite_db.messages m
@@ -321,7 +339,8 @@ func buildCache(dbPath, analyticsDir string, fullRebuild bool) (*buildResult, er
 			id,
 			COALESCE(TRY_CAST(email_address AS VARCHAR), '') as email_address,
 			COALESCE(TRY_CAST(domain AS VARCHAR), '') as domain,
-			COALESCE(TRY_CAST(display_name AS VARCHAR), '') as display_name
+			COALESCE(TRY_CAST(display_name AS VARCHAR), '') as display_name,
+			COALESCE(TRY_CAST(phone_number AS VARCHAR), '') as phone_number
 		FROM sqlite_db.participants
 	) TO '%s/participants.parquet' (
 		FORMAT PARQUET,
@@ -372,7 +391,8 @@ func buildCache(dbPath, analyticsDir string, fullRebuild bool) (*buildResult, er
 	COPY (
 		SELECT
 			id,
-			COALESCE(TRY_CAST(source_conversation_id AS VARCHAR), '') as source_conversation_id
+			COALESCE(TRY_CAST(source_conversation_id AS VARCHAR), '') as source_conversation_id,
+			COALESCE(TRY_CAST(title AS VARCHAR), '') as title
 		FROM sqlite_db.conversations
 	) TO '%s/conversations.parquet' (
 		FORMAT PARQUET,
@@ -391,10 +411,11 @@ func buildCache(dbPath, analyticsDir string, fullRebuild bool) (*buildResult, er
 		exportedCount = 0
 	}
 
-	// Save sync state
+	// Save sync state with schema version for compatibility detection.
 	state := syncState{
 		LastMessageID: maxID,
 		LastSyncAt:    time.Now(),
+		SchemaVersion: cacheSchemaVersion,
 	}
 	stateData, _ := json.Marshal(state)
 	if err := os.WriteFile(stateFile, stateData, 0644); err != nil {
@@ -592,15 +613,15 @@ func setupSQLiteSource(duckDB *sql.DB, dbPath string) (cleanup func(), err error
 		query         string
 		typeOverrides string // DuckDB types parameter for read_csv_auto (empty = infer all)
 	}{
-		{"messages", "SELECT id, source_id, source_message_id, conversation_id, subject, snippet, sent_at, size_estimate, has_attachments, deleted_from_source_at FROM messages WHERE sent_at IS NOT NULL",
+		{"messages", "SELECT id, source_id, source_message_id, conversation_id, subject, snippet, sent_at, size_estimate, has_attachments, attachment_count, deleted_from_source_at, sender_id, message_type FROM messages WHERE sent_at IS NOT NULL",
 			"types={'sent_at': 'TIMESTAMP', 'deleted_from_source_at': 'TIMESTAMP'}"},
 		{"message_recipients", "SELECT message_id, participant_id, recipient_type, display_name FROM message_recipients", ""},
 		{"message_labels", "SELECT message_id, label_id FROM message_labels", ""},
 		{"attachments", "SELECT message_id, size, filename FROM attachments", ""},
-		{"participants", "SELECT id, email_address, domain, display_name FROM participants", ""},
+		{"participants", "SELECT id, email_address, domain, display_name, phone_number FROM participants", ""},
 		{"labels", "SELECT id, name FROM labels", ""},
 		{"sources", "SELECT id, identifier FROM sources", ""},
-		{"conversations", "SELECT id, source_conversation_id FROM conversations", ""},
+		{"conversations", "SELECT id, source_conversation_id, title FROM conversations", ""},
 	}
 
 	for _, t := range tables {

cmd/msgvault/cmd/import.go

@@ -0,0 +1,236 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/spf13/cobra"
+	"github.com/wesm/msgvault/internal/store"
+	"github.com/wesm/msgvault/internal/whatsapp"
+)
+
+var (
+	importType        string
+	importPhone       string
+	importMediaDir    string
+	importContacts    string
+	importLimit       int
+	importDisplayName string
+)
+
+var importCmd = &cobra.Command{
+	Use:   "import [path]",
+	Short: "Import messages from external sources",
+	Long: `Import messages from external message databases.
+
+Currently supported types:
+  whatsapp    Import from a decrypted WhatsApp msgstore.db
+
+Examples:
+  msgvault import --type whatsapp --phone "+447700900000" /path/to/msgstore.db
+  msgvault import --type whatsapp --phone "+447700900000" --contacts ~/contacts.vcf /path/to/msgstore.db
+  msgvault import --type whatsapp --phone "+447700900000" --media-dir /path/to/Media /path/to/msgstore.db`,
+	Args: cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		if err := MustBeLocal("import"); err != nil {
+			return err
+		}
+
+		sourcePath := args[0]
+
+		// Validate source file exists.
+		if _, err := os.Stat(sourcePath); err != nil {
+			return fmt.Errorf("source file not found: %w", err)
+		}
+
+		switch strings.ToLower(importType) {
+		case "whatsapp":
+			return runWhatsAppImport(cmd, sourcePath)
+		default:
+			return fmt.Errorf("unsupported import type %q (supported: whatsapp)", importType)
+		}
+	},
+}
+
+func runWhatsAppImport(cmd *cobra.Command, sourcePath string) error {
+	// Validate phone number.
+	if importPhone == "" {
+		return fmt.Errorf("--phone is required for WhatsApp import (E.164 format, e.g., +447700900000)")
+	}
+	if !strings.HasPrefix(importPhone, "+") {
+		return fmt.Errorf("phone number must be in E.164 format (starting with +), got %q", importPhone)
+	}
+
+	// Validate media dir if provided.
+	if importMediaDir != "" {
+		if info, err := os.Stat(importMediaDir); err != nil || !info.IsDir() {
+			return fmt.Errorf("media directory not found or not a directory: %s", importMediaDir)
+		}
+	}
+
+	// Open database.
+	dbPath := cfg.DatabaseDSN()
+	s, err := store.Open(dbPath)
+	if err != nil {
+		return fmt.Errorf("open database: %w", err)
+	}
+	defer s.Close()
+
+	if err := s.InitSchema(); err != nil {
+		return fmt.Errorf("init schema: %w", err)
+	}
+
+	// Set up context with cancellation.
+	ctx, cancel := context.WithCancel(cmd.Context())
+	defer cancel()
+
+	// Handle Ctrl+C gracefully.
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-sigChan
+		fmt.Println("\nInterrupted. Saving checkpoint...")
+		cancel()
+	}()
+
+	// Build import options.
+	opts := whatsapp.DefaultOptions()
+	opts.Phone = importPhone
+	opts.DisplayName = importDisplayName
+	opts.MediaDir = importMediaDir
+	opts.AttachmentsDir = cfg.AttachmentsDir()
+	opts.Limit = importLimit
+
+	// Create importer with CLI progress.
+	progress := &ImportCLIProgress{}
+	importer := whatsapp.NewImporter(s, progress)
+
+	fmt.Printf("Importing WhatsApp messages from %s\n", sourcePath)
+	fmt.Printf("Phone: %s\n", importPhone)
+	if importMediaDir != "" {
+		fmt.Printf("Media: %s\n", importMediaDir)
+	}
+	if importLimit > 0 {
+		fmt.Printf("Limit: %d messages\n", importLimit)
+	}
+	fmt.Println()
+
+	summary, err := importer.Import(ctx, sourcePath, opts)
+	if err != nil {
+		if ctx.Err() != nil {
+			fmt.Println("\nImport interrupted. Run again to continue.")
+			return nil
+		}
+		return fmt.Errorf("import failed: %w", err)
+	}
+
+	// Import contacts if provided.
+	if importContacts != "" {
+		fmt.Printf("\nImporting contacts from %s...\n", importContacts)
+		matched, total, err := whatsapp.ImportContacts(s, importContacts)
+		if err != nil {
+			fmt.Printf("Warning: contact import failed: %v\n", err)
+		} else {
+			fmt.Printf("  Contacts: %d in file, %d phone numbers matched to participants\n", total, matched)
+		}
+	}
+
+	// Print summary.
+	fmt.Println()
+	fmt.Println("Import complete!")
+	fmt.Printf("  Duration:       %s\n", summary.Duration.Round(time.Second))
+	fmt.Printf("  Chats:          %d\n", summary.ChatsProcessed)
+	fmt.Printf("  Messages:       %d processed, %d added, %d skipped\n",
+		summary.MessagesProcessed, summary.MessagesAdded, summary.MessagesSkipped)
+	fmt.Printf("  Participants:   %d\n", summary.Participants)
+	fmt.Printf("  Reactions:      %d\n", summary.ReactionsAdded)
+	fmt.Printf("  Attachments:    %d found", summary.AttachmentsFound)
+	if summary.MediaCopied > 0 {
+		fmt.Printf(", %d files copied", summary.MediaCopied)
+	}
+	fmt.Println()
+	if summary.Errors > 0 {
+		fmt.Printf("  Errors:         %d\n", summary.Errors)
+	}
+
+	if summary.MessagesAdded > 0 {
+		rate := float64(summary.MessagesAdded) / summary.Duration.Seconds()
+		fmt.Printf("  Rate:           %.0f messages/sec\n", rate)
+	}
+
+	return nil
+}
+
+// ImportCLIProgress implements whatsapp.ImportProgress for terminal output.
+type ImportCLIProgress struct {
+	startTime time.Time
+	lastPrint time.Time
+	currentChat string
+}
+
+func (p *ImportCLIProgress) OnStart() {
+	p.startTime = time.Now()
+	p.lastPrint = time.Now()
+}
+
+func (p *ImportCLIProgress) OnChatStart(chatJID, chatTitle string, messageCount int) {
+	p.currentChat = chatTitle
+	// Don't print every chat start — too noisy for 13k+ chats.
+}
+
+func (p *ImportCLIProgress) OnProgress(processed, added, skipped int64) {
+	// Throttle output to every 2 seconds.
+	if time.Since(p.lastPrint) < 2*time.Second {
+		return
+	}
+	p.lastPrint = time.Now()
+
+	elapsed := time.Since(p.startTime)
+	rate := 0.0
+	if elapsed.Seconds() >= 1 {
+		rate = float64(added) / elapsed.Seconds()
+	}
+
+	elapsedStr := formatDuration(elapsed)
+
+	chatStr := ""
+	if p.currentChat != "" {
+		// Truncate long chat names.
+		name := p.currentChat
+		if len(name) > 30 {
+			name = name[:27] + "..."
+		}
+		chatStr = fmt.Sprintf(" | Chat: %s", name)
+	}
+
+	fmt.Printf("\r  Processed: %d | Added: %d | Skipped: %d | Rate: %.0f/s | Elapsed: %s%s    ",
+		processed, added, skipped, rate, elapsedStr, chatStr)
+}
+
+func (p *ImportCLIProgress) OnChatComplete(chatJID string, messagesAdded int64) {
+	// Quiet — progress line shows the aggregate.
+}
+
+func (p *ImportCLIProgress) OnComplete(summary *whatsapp.ImportSummary) {
+	fmt.Println() // Clear the progress line.
+}
+
+func (p *ImportCLIProgress) OnError(err error) {
+	fmt.Printf("\nWarning: %v\n", err)
+}
+
+func init() {
+	importCmd.Flags().StringVar(&importType, "type", "", "import source type (required: whatsapp)")
+	importCmd.Flags().StringVar(&importPhone, "phone", "", "your phone number in E.164 format (required for whatsapp)")
+	importCmd.Flags().StringVar(&importMediaDir, "media-dir", "", "path to decrypted Media folder (optional)")
+	importCmd.Flags().StringVar(&importContacts, "contacts", "", "path to contacts .vcf file for name resolution (optional)")
+	importCmd.Flags().IntVar(&importLimit, "limit", 0, "limit number of messages (for testing)")
+	importCmd.Flags().StringVar(&importDisplayName, "display-name", "", "display name for the phone owner")
+	_ = importCmd.MarkFlagRequired("type")
+	rootCmd.AddCommand(importCmd)
+}