Add asynchronous historical import jobs

api/openapi.yaml

@@ -4829,6 +4829,106 @@ components:
       required:
         - identifier
       type: object
+    ImportJobRequest:
+      additionalProperties: false
+      properties:
+        account:
+          minLength: 1
+          type: string
+        after:
+          pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}$
+          type: string
+        before:
+          pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}$
+          type: string
+        limit:
+          format: int64
+          minimum: 0
+          type: integer
+        noresume:
+          type: boolean
+        query:
+          description: Gmail search query; not supported for IMAP sources
+          type: string
+      required:
+        - account
+      type: object
+    ImportJobResponse:
+      additionalProperties: true
+      properties:
+        account:
+          type: string
+        added:
+          format: int64
+          type: integer
+        created_at:
+          format: date-time
+          type: string
+        error:
+          type: string
+        finished_at:
+          format: date-time
+          type:
+            - string
+            - "null"
+        job_id:
+          type: string
+        processed:
+          format: int64
+          type: integer
+        skipped:
+          format: int64
+          type: integer
+        started_at:
+          format: date-time
+          type:
+            - string
+            - "null"
+        status:
+          enum:
+            - pending
+            - running
+            - done
+            - failed
+          type: string
+        summary:
+          $ref: "#/components/schemas/ImportJobSummary"
+      required:
+        - job_id
+        - account
+        - status
+        - processed
+        - added
+        - skipped
+        - created_at
+        - started_at
+        - finished_at
+      type: object
+    ImportJobSummary:
+      additionalProperties: true
+      properties:
+        added:
+          format: int64
+          type: integer
+        errors:
+          format: int64
+          type: integer
+        processed:
+          format: int64
+          type: integer
+        skipped:
+          format: int64
+          type: integer
+        updated:
+          format: int64
+          type: integer
+      required:
+        - processed
+        - added
+        - updated
+        - skipped
+        - errors
+      type: object
     ImportRequest:
       additionalProperties: false
       properties:
@@ -11251,7 +11351,7 @@ components:
       type: apiKey
 info:
   title: msgvault API
-  version: 2.13.0
+  version: 2.14.0
 openapi: 3.1.0
 paths:
   /api/ping:
@@ -16318,6 +16418,127 @@ paths:
       summary: Import one meeting
       tags:
         - API
+  /api/v1/imports:
+    post:
+      operationId: createImportJob
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/ImportJobRequest"
+        required: true
+      responses:
+        "202":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ImportJobResponse"
+          description: Accepted
+        "400":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "401":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "404":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "409":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "413":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "415":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "422":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "500":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "503":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        default:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+      security:
+        - apiKey: []
+      summary: Start a bounded historical import
+      tags:
+        - API
+  /api/v1/imports/{job_id}:
+    get:
+      operationId: getImportJob
+      parameters:
+        - description: Historical import job ID
+          in: path
+          name: job_id
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ImportJobResponse"
+          description: OK
+        "401":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        "404":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+        default:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+          description: Error
+      security:
+        - apiKey: []
+      summary: Get historical import status
+      tags:
+        - API
   /api/v1/integrations/tasks/search:
     get:
       operationId: searchIntegrationTasks

cmd/msgvault/cmd/cache_staleness_meeting_refresh_test.go

@@ -133,7 +133,7 @@ func TestCacheNeedsBuild_CirclebackRefresh(t *testing.T) {
 	assert.Equal("Refreshed Meeting", cachedSubject)
 }
 
-func TestCacheNeedsBuild_SupersededCirclebackRunWithoutCheckpoint(t *testing.T) {
+func TestCacheNeedsBuild_FailedCirclebackRunWithoutCheckpoint(t *testing.T) {
 	require := require.New(t)
 	assert := assert.New(t)
 	tmp := t.TempDir()
@@ -169,8 +169,7 @@ func TestCacheNeedsBuild_SupersededCirclebackRunWithoutCheckpoint(t *testing.T)
 		WHERE source_message_id = 'meeting:failed-refresh-1'
 	`)
 	require.NoError(err)
-	_, err = st.StartSync(first.SourceID, circleback.SourceType)
-	require.NoError(err, "starting the replacement run supersedes the abandoned run")
+	require.NoError(st.FailSync(abandonedRunID, "worker stopped"))
 
 	var status string
 	var additions, updates int64

cmd/msgvault/cmd/remove_account_test.go

@@ -570,7 +570,7 @@ func TestRemoveAccountCmd_SkipsDeletionWhenRemovedAccountHasActiveSync(t *testin
 	// RemoveSource cascades away.
 	_, err = s.StartSync(aliceSrc.ID, "full")
 	require.NoError(err, "StartSync")
-	_ = s.Close()
+	t.Cleanup(func() { _ = s.Close() })
 
 	filePath := seedAttachmentFile(t, attachmentsDir, "dd/hashA", "content-a")
 
@@ -606,7 +606,7 @@ func TestRemoveAccountConfirmedDoesNotBypassActiveSyncGuard(t *testing.T) {
 	require.NoError(err, "GetSourceByIdentifier")
 	_, err = s.StartSync(aliceSrc.ID, "full")
 	require.NoError(err, "StartSync")
-	_ = s.Close()
+	t.Cleanup(func() { _ = s.Close() })
 
 	savedCfg := cfg
 	defer func() { cfg = savedCfg }()
@@ -1205,7 +1205,7 @@ func TestRemoveAccountCmd_DiscordPreservesTokenDuringActiveSync(t *testing.T) {
 	require.NoError(err)
 	_, err = st.StartSync(source.ID, "discord")
 	require.NoError(err)
-	require.NoError(st.Close())
+	t.Cleanup(func() { _ = st.Close() })
 
 	manager := discord.NewTokenManager(filepath.Join(tmpDir, "tokens"))
 	require.NoError(manager.Save(discord.NewTokenRecord(

cmd/msgvault/cmd/serve.go

@@ -242,6 +242,14 @@ func runServe(cmd *cobra.Command, args []string) error {
 	if err := s.InitSchemaContext(cmd.Context()); err != nil {
 		return fmt.Errorf("init schema: %w", err)
 	}
+	failedUnfinishedImports, err := s.FailUnfinishedSyncOperationsContext(cmd.Context())
+	if err != nil {
+		return fmt.Errorf("recover unfinished historical imports: %w", err)
+	}
+	if failedUnfinishedImports > 0 {
+		logger.Warn("marked historical imports abandoned by the previous daemon as failed",
+			"count", failedUnfinishedImports)
+	}
 	logger.Info("daemon startup step complete", "step", "init_archive_schema")
 	// Legacy [identity] migration is deferred to the first scheduled sync's
 	// runPostSourceCreateMigrations call, which fires AFTER that sync's
@@ -1457,7 +1465,30 @@ func (a *storeAPIAdapter) RunCLISync(
 	req api.CLISyncRequest,
 	emit func(api.CLISyncEvent) error,
 ) error {
-	return a.runCLISyncWithRunner(ctx, req, emit, runDaemonCLISubprocessStream)
+	return a.runCLISyncOperationWithRunner(ctx, req, emit, runDaemonCLISubprocessStream)
+}
+
+func (a *storeAPIAdapter) runCLISyncOperationWithRunner(
+	ctx context.Context,
+	req api.CLISyncRequest,
+	emit func(api.CLISyncEvent) error,
+	run cliSyncSubprocessRunner,
+) error {
+	err := a.runCLISyncWithRunner(ctx, req, emit, run)
+	if err == nil && ctx.Err() != nil {
+		err = ctx.Err()
+	}
+	if req.OperationID == "" {
+		return err
+	}
+	status := "done"
+	if err != nil {
+		status = "failed"
+	}
+	if finishErr := a.store.FinishSyncOperation(req.OperationID, status); finishErr != nil {
+		return errors.Join(err, fmt.Errorf("finish sync operation: %w", finishErr))
+	}
+	return err
 }
 
 type cliSyncSubprocessRunner func(
@@ -1505,6 +1536,9 @@ func cliSyncSubprocessArgs(req api.CLISyncRequest) []string {
 		if req.SourceIDSet {
 			args = append(args, "--source-id", strconv.FormatInt(req.SourceID, 10))
 		}
+		if req.OperationID != "" {
+			args = append(args, "--sync-operation-id", req.OperationID)
+		}
 		if req.Query != "" {
 			args = append(args, "--query", req.Query)
 		}
@@ -2732,6 +2766,14 @@ func (a *storeAPIAdapter) GetLatestSync(sourceID int64) (*store.SyncRun, error)
 	return a.store.GetLatestSync(sourceID)
 }
 
+func (a *storeAPIAdapter) GetSyncOperation(operationID string) (*store.SyncOperation, error) {
+	return a.store.GetSyncOperation(operationID)
+}
+
+func (a *storeAPIAdapter) CreateSyncOperation(sourceID int64, operationID string) (*store.SyncOperation, error) {
+	return a.store.CreateSyncOperation(sourceID, operationID)
+}
+
 func (a *storeAPIAdapter) GetLastSuccessfulSync(sourceID int64) (*store.SyncRun, error) {
 	return a.store.GetLastSuccessfulSync(sourceID)
 }