feat: add generic single-meeting ingestion endpoint

api/openapi.yaml

@@ -2626,6 +2626,134 @@ components:
         strongest_excerpt:
           type: string
       type: object
+    Meeting:
+      additionalProperties: false
+      allOf:
+        - anyOf:
+            - properties:
+                summary_markdown:
+                  minLength: 1
+                  type: string
+              required:
+                - summary_markdown
+              type: object
+            - properties:
+                summary_text:
+                  minLength: 1
+                  type: string
+              required:
+                - summary_text
+              type: object
+            - properties:
+                transcript:
+                  minLength: 1
+                  type: string
+              required:
+                - transcript
+              type: object
+            - properties:
+                transcript_segments:
+                  minItems: 1
+                  type: array
+              required:
+                - transcript_segments
+              type: object
+        - not:
+            properties:
+              transcript:
+                minLength: 1
+                type: string
+              transcript_segments:
+                minItems: 1
+                type: array
+            required:
+              - transcript
+              - transcript_segments
+            type: object
+      properties:
+        attendees:
+          items:
+            $ref: "#/components/schemas/MeetingPerson"
+          type:
+            - array
+            - "null"
+        ended_at:
+          format: date-time
+          type: string
+        external_id:
+          maxLength: 256
+          type: string
+        metadata:
+          additionalProperties: {}
+          type: object
+        organizer:
+          $ref: "#/components/schemas/MeetingPerson"
+        started_at:
+          format: date-time
+          type: string
+        summary_markdown:
+          type: string
+        summary_text:
+          type: string
+        title:
+          maxLength: 4096
+          type: string
+        transcript:
+          type: string
+        transcript_segments:
+          items:
+            $ref: "#/components/schemas/TranscriptSegment"
+          type:
+            - array
+            - "null"
+      required:
+        - external_id
+        - started_at
+      type: object
+    MeetingImportRequest:
+      additionalProperties: false
+      properties:
+        meeting:
+          $ref: "#/components/schemas/Meeting"
+        source:
+          $ref: "#/components/schemas/Source"
+      required:
+        - source
+        - meeting
+      type: object
+    MeetingImportResponse:
+      additionalProperties: true
+      properties:
+        message_id:
+          format: int64
+          type: integer
+        source_id:
+          format: int64
+          type: integer
+        source_message_id:
+          type: string
+        status:
+          enum:
+            - created
+            - updated
+          type: string
+      required:
+        - status
+        - source_id
+        - message_id
+        - source_message_id
+      type: object
+    MeetingPerson:
+      additionalProperties: false
+      properties:
+        email:
+          format: email
+          type: string
+        name:
+          type: string
+      required:
+        - email
+      type: object
     MessageDetail:
       additionalProperties: true
       properties:
@@ -3798,6 +3926,22 @@ components:
         - generation
         - messages
       type: object
+    Source:
+      additionalProperties: false
+      properties:
+        account_email:
+          format: email
+          type: string
+        display_name:
+          maxLength: 256
+          type: string
+        identifier:
+          maxLength: 128
+          type: string
+      required:
+        - identifier
+        - account_email
+      type: object
     SourceCount:
       additionalProperties: true
       properties:
@@ -4506,6 +4650,21 @@ components:
         - label_count
         - account_count
       type: object
+    TranscriptSegment:
+      additionalProperties: false
+      properties:
+        offset_seconds:
+          format: double
+          minimum: 0
+          type: number
+        speaker:
+          type: string
+        text:
+          type: string
+      required:
+        - speaker
+        - text
+      type: object
     UpdateRequest:
       additionalProperties: false
       properties:
@@ -4545,7 +4704,7 @@ components:
       type: apiKey
 info:
   title: msgvault API
-  version: 1.32.0
+  version: 1.33.0
 openapi: 3.1.0
 paths:
   /api/ping:
@@ -7263,6 +7422,39 @@ paths:
       summary: Remove a link edge between two participants
       tags:
         - API
+  /api/v1/import/meeting:
+    post:
+      operationId: importMeeting
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/MeetingImportRequest"
+        required: true
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/MeetingImportResponse"
+          description: OK
+        "201":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/MeetingImportResponse"
+          description: Created
+        default:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+      security:
+        - apiKey: []
+      summary: Import one meeting
+      tags:
+        - API
   /api/v1/integrations/tasks/search:
     get:
       operationId: searchIntegrationTasks

cmd/msgvault/cmd/build_cache.go

@@ -735,6 +735,18 @@ func buildCacheLocked(
 		return nil, fmt.Errorf("inspect attachment MIME schema: %w", err)
 	}
 	sourceSnapshot.hasAttachmentMIME = attachmentMIMEColumnCount > 0
+	var messageSourceAttributionColumnCount int
+	if err := sourceSnapshot.QueryRow(`
+		SELECT COUNT(*) FROM pragma_table_info('messages')
+		WHERE name = 'source_is_from_me'
+	`).Scan(&messageSourceAttributionColumnCount); err != nil {
+		return nil, fmt.Errorf("inspect message attribution schema: %w", err)
+	}
+	messageSourceAttribution := "COALESCE(m.is_from_me, FALSE)"
+	if messageSourceAttributionColumnCount > 0 {
+		messageSourceAttribution = "COALESCE(m.source_is_from_me, FALSE)"
+	}
+	sourceSnapshot.hasMessageSourceAttribution = messageSourceAttributionColumnCount > 0
 	if err := sourceSnapshot.Prepare(); err != nil {
 		return nil, err
 	}
@@ -1067,7 +1079,7 @@ func buildCacheLocked(
 			m.deleted_from_source_at,
 			m.sender_id,
 			COALESCE(TRY_CAST(m.message_type AS VARCHAR), '') as message_type,
-			(COALESCE(m.is_from_me, FALSE) OR EXISTS (
+			(%s OR EXISTS (
 				SELECT 1 FROM sqlite_db.account_identities ai
 				JOIN sqlite_db.participants sp ON sp.id = m.sender_id
 				WHERE ai.source_id = m.source_id
@@ -1090,7 +1102,7 @@ func buildCacheLocked(
 		OVERWRITE_OR_IGNORE,
 		COMPRESSION 'zstd'
 	)
-	`, idFilter, escapedMessagesDir)); err != nil {
+	`, messageSourceAttribution, idFilter, escapedMessagesDir)); err != nil {
 		return nil, fmt.Errorf("export messages: %w", err)
 	}
 
@@ -1126,7 +1138,7 @@ func buildCacheLocked(
 				m.deleted_from_source_at,
 				m.sender_id,
 				COALESCE(TRY_CAST(m.message_type AS VARCHAR), '') as message_type,
-				(COALESCE(m.is_from_me, FALSE) OR EXISTS (
+				(%s OR EXISTS (
 					SELECT 1 FROM sqlite_db.account_identities ai
 					JOIN sqlite_db.participants sp ON sp.id = m.sender_id
 					WHERE ai.source_id = m.source_id
@@ -1143,7 +1155,7 @@ func buildCacheLocked(
 			FROM sqlite_db.messages m
 			WHERE 1 = 0
 		) TO '%s' (FORMAT PARQUET, COMPRESSION 'zstd')
-		`, escapedEmptyShard)); err != nil {
+		`, messageSourceAttribution, escapedEmptyShard)); err != nil {
 			return nil, fmt.Errorf("export empty messages shard: %w", err)
 		}
 	}
@@ -1448,12 +1460,13 @@ func writeCacheStatsLine(w io.Writer, format string, args ...any) error {
 // transaction directly. The fallback exports every table from one go-sqlite3
 // transaction before DuckDB reads the resulting static CSV files.
 type cacheSourceSnapshot struct {
-	duckDB            *sql.DB
-	duckTx            *sql.Tx
-	sqliteDB          *sql.DB
-	sqliteTx          *sql.Tx
-	tmpDir            string
-	hasAttachmentMIME bool
+	duckDB                      *sql.DB
+	duckTx                      *sql.Tx
+	sqliteDB                    *sql.DB
+	sqliteTx                    *sql.Tx
+	tmpDir                      string
+	hasAttachmentMIME           bool
+	hasMessageSourceAttribution bool
 }
 
 type cacheSnapshotTable struct {
@@ -1574,6 +1587,14 @@ func (s *cacheSourceSnapshot) tables() []cacheSnapshotTable {
 	if s.hasAttachmentMIME {
 		attachmentQuery = "SELECT id, message_id, size, filename, mime_type FROM attachments"
 	}
+	messageColumns := "id, source_id, source_message_id, conversation_id, subject, snippet, sent_at, size_estimate, has_attachments, attachment_count, deleted_from_source_at, deleted_at, sender_id, message_type, is_from_me"
+	messageTypes := "types={'id': 'BIGINT', 'source_id': 'BIGINT', 'source_message_id': 'VARCHAR', 'conversation_id': 'BIGINT', 'subject': 'VARCHAR', 'snippet': 'VARCHAR', 'sent_at': 'TIMESTAMP', 'size_estimate': 'BIGINT', 'has_attachments': 'BOOLEAN', 'attachment_count': 'INTEGER', 'deleted_from_source_at': 'TIMESTAMP', 'deleted_at': 'TIMESTAMP', 'sender_id': 'BIGINT', 'message_type': 'VARCHAR', 'is_from_me': 'BOOLEAN'"
+	if s.hasMessageSourceAttribution {
+		messageColumns += ", source_is_from_me"
+		messageTypes += ", 'source_is_from_me': 'BOOLEAN'"
+	}
+	messageTypes += "}"
+
 	// Every column carries an explicit type so the schema DuckDB sees never
 	// depends on the data: read_csv_auto sniffs empty or all-NULL columns as
 	// VARCHAR, which breaks downstream SQL that binds these columns against
@@ -1583,8 +1604,7 @@ func (s *cacheSourceSnapshot) tables() []cacheSnapshotTable {
 		// `deleted_at IS NULL` filter on this path the same way it does
 		// on the sqlite_scanner path; otherwise DuckDB binds against a
 		// CSV view that lacks the column and the export fails on Windows.
-		{tableMessages, "SELECT id, source_id, source_message_id, conversation_id, subject, snippet, sent_at, size_estimate, has_attachments, attachment_count, deleted_from_source_at, deleted_at, sender_id, message_type, is_from_me FROM messages WHERE sent_at IS NOT NULL",
-			"types={'id': 'BIGINT', 'source_id': 'BIGINT', 'source_message_id': 'VARCHAR', 'conversation_id': 'BIGINT', 'subject': 'VARCHAR', 'snippet': 'VARCHAR', 'sent_at': 'TIMESTAMP', 'size_estimate': 'BIGINT', 'has_attachments': 'BOOLEAN', 'attachment_count': 'INTEGER', 'deleted_from_source_at': 'TIMESTAMP', 'deleted_at': 'TIMESTAMP', 'sender_id': 'BIGINT', 'message_type': 'VARCHAR', 'is_from_me': 'BOOLEAN'}"},
+		{tableMessages, "SELECT " + messageColumns + " FROM messages WHERE sent_at IS NOT NULL", messageTypes},
 		{"message_recipients", "SELECT message_id, participant_id, recipient_type, display_name FROM message_recipients",
 			"types={'message_id': 'BIGINT', 'participant_id': 'BIGINT', 'recipient_type': 'VARCHAR', 'display_name': 'VARCHAR'}"},
 		{"message_labels", "SELECT message_id, label_id FROM message_labels",

cmd/msgvault/cmd/build_cache_identity_test.go

@@ -129,6 +129,14 @@ func TestBuildCache_DerivesIsFromMeAndIdentityDatasets(t *testing.T) {
 	})
 	require.NoError(err)
 	require.NoError(st.ReplaceMessageRecipients(controlMsgID, "from", []int64{otherParticipantID}, []string{""}))
+	_, err = st.DB().Exec(st.Rebind(`
+		UPDATE messages
+		SET is_from_me = TRUE,
+		    source_is_from_me = FALSE,
+		    identity_is_from_me = TRUE
+		WHERE id = ?
+	`), controlMsgID)
+	require.NoError(err, "simulate stale persisted effective attribution")
 
 	require.NoError(st.Close())
 

cmd/msgvault/cmd/build_cache_test.go

@@ -2491,6 +2491,42 @@ func TestBuildCacheCSVSnapshotFallback(t *testing.T) {
 		assert.Positive(t, result.ExportedCount)
 	})
 
+	t.Run("message attribution provenance", func(t *testing.T) {
+		require := require.New(t)
+		assert := assert.New(t)
+		tmpDir := setupTestSQLite(t)
+		dbPath := filepath.Join(tmpDir, "test.db")
+		analyticsDir := filepath.Join(tmpDir, "analytics")
+
+		db, err := sql.Open("sqlite3", dbPath)
+		require.NoError(err)
+		_, err = db.Exec(`
+			ALTER TABLE messages ADD COLUMN source_is_from_me BOOLEAN DEFAULT FALSE;
+			UPDATE messages
+			SET is_from_me = FALSE, source_is_from_me = TRUE
+			WHERE id = 1;
+		`)
+		require.NoError(err)
+		require.NoError(db.Close())
+
+		result, err := buildCache(dbPath, analyticsDir, true)
+		require.NoError(err)
+		assert.Positive(result.ExportedCount)
+
+		duckDB, err := sql.Open("duckdb", "")
+		require.NoError(err)
+		defer func() { require.NoError(duckDB.Close()) }()
+
+		var isFromMe bool
+		require.NoError(duckDB.QueryRow(
+			`SELECT is_from_me
+			 FROM read_parquet(?, hive_partitioning=true)
+			 WHERE id = 1`,
+			filepath.Join(analyticsDir, "messages", "**", "*.parquet"),
+		).Scan(&isFromMe))
+		assert.True(isFromMe, "CSV fallback must export source-native attribution")
+	})
+
 	t.Run("empty", func(t *testing.T) {
 		tmpDir := setupTestSQLiteEmpty(t)
 		result, err := buildCache(filepath.Join(tmpDir, "test.db"), filepath.Join(tmpDir, "analytics"), true)