package backend

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/pathutil"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
)

type staticImageExtractor struct {
	text string
}

// fakePNG and fakePNGAlt are fake image payloads prefixed with the PNG magic
// bytes so they pass the raster-image sniff in the extraction path.
var (
	fakePNG    = append([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1A, '\n'}, "fake-png"...)
	fakePNGAlt = append([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1A, '\n'}, "first-image-bytes"...)
)

func (e *staticImageExtractor) ExtractImageText(ctx context.Context, req ImageExtractRequest) (string, ImageExtractUsage, error) {
	return e.text, ImageExtractUsage{}, nil
}

type gatedImageExtractor struct {
	started chan struct{}
	release chan struct{}

	mu    sync.Mutex
	calls int
}

func (e *gatedImageExtractor) ExtractImageText(ctx context.Context, req ImageExtractRequest) (string, ImageExtractUsage, error) {
	e.mu.Lock()
	e.calls++
	call := e.calls
	e.mu.Unlock()
	if call == 1 {
		select {
		case e.started <- struct{}{}:
		default:
		}
		select {
		case <-e.release:
		case <-ctx.Done():
			return "", ImageExtractUsage{}, ctx.Err()
		}
		return "old caption", ImageExtractUsage{}, nil
	}
	return "new caption", ImageExtractUsage{}, nil
}

func TestAsyncImageExtractUpdatesContentText(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})

	if _, err := b.Write("/img/a.png", fakePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	// Durable delivery: the write enqueues an img_extract_text task, but the
	// semantic worker is not running in backend tests. Call the processing
	// function directly as the worker would.
	fileID, _, _, _ := mustFileForPath(t, b, "/img/a.png")
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/a.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/a.png", 2*time.Second)
	if !strings.Contains(got, "cat on sofa") {
		t.Fatalf("unexpected extracted text: %q", got)
	}
}

func TestAsyncImageExtractAutoEmbeddingUpdatesContentTextWithoutSemanticTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/auto.png", "image/png", fakePNG)
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/auto.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/auto.png", time.Second)
	if !strings.Contains(got, "cat on sofa") {
		t.Fatalf("unexpected extracted text: %q", got)
	}

	if tasks := loadSemanticTasksForFile(t, b, fileID); len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestAsyncImageExtractSkipsStaleChecksum(t *testing.T) {
	extractor := &gatedImageExtractor{
		started: make(chan struct{}, 1),
		release: make(chan struct{}),
	}
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: extractor,
		},
	})

	if _, err := b.Write("/img/b.png", fakePNGAlt, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/img/b.png")

	// Start extraction for revision 1 (gated — blocks until released).
	resultCh := make(chan ImageExtractResult, 1)
	errCh := make(chan error, 1)
	go func() {
		result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
			FileID:      fileID,
			Path:        "/img/b.png",
			ContentType: "image/png",
			Revision:    1,
		})
		resultCh <- result
		errCh <- err
	}()
	select {
	case <-extractor.started:
	case <-time.After(2 * time.Second):
		t.Fatal("first extraction did not start")
	}

	// Overwrite to revision 2 while revision 1 extraction is in-flight.
	if _, err := b.Write("/img/b.png", fakePNG, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	close(extractor.release)

	// The stale revision 1 extraction should complete without writing.
	if gotErr := <-errCh; gotErr != nil {
		t.Fatal(gotErr)
	}
	if got := <-resultCh; got != ImageExtractResultStale {
		t.Fatalf("stale result=%q, want %q", got, ImageExtractResultStale)
	}

	// Now process revision 2 directly.
	_, revision, _, _ := mustFileForPath(t, b, "/img/b.png")
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/b.png",
		ContentType: "image/png",
		Revision:    revision,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/b.png", 3*time.Second)
	if got != "new caption" {
		t.Fatalf("expected final extracted text %q, got %q", "new caption", got)
	}
}

func TestAsyncImageExtractAutoEmbeddingStaleResultDoesNotQueueOrOverwriteCurrentText(t *testing.T) {
	extractor := &gatedImageExtractor{
		started: make(chan struct{}, 1),
		release: make(chan struct{}),
	}
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: extractor,
		},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/auto-stale.png", "image/png", fakePNGAlt)
	resultCh := make(chan ImageExtractResult, 1)
	errCh := make(chan error, 1)
	go func() {
		result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
			FileID:      fileID,
			Path:        "/img/auto-stale.png",
			ContentType: "image/png",
			Revision:    1,
		})
		resultCh <- result
		errCh <- err
	}()
	select {
	case <-extractor.started:
	case <-time.After(2 * time.Second):
		t.Fatal("first extraction did not start")
	}

	if _, err := b.Write("/img/auto-stale.png", fakePNG, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/auto-stale.png",
		ContentType: "image/png",
		Revision:    2,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}
	close(extractor.release)
	if gotErr := <-errCh; gotErr != nil {
		t.Fatal(gotErr)
	}
	if got := <-resultCh; got != ImageExtractResultStale {
		t.Fatalf("stale result=%q, want %q", got, ImageExtractResultStale)
	}

	got := waitForContentText(t, b, "/img/auto-stale.png", time.Second)
	if got != "new caption" {
		t.Fatalf("expected final extracted text %q, got %q", "new caption", got)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != "queued" || tasks[0].ResourceVersion != 2 {
		t.Fatalf("unexpected semantic task history: %+v", tasks)
	}
}

func TestAsyncImageExtractQueuesEmbedTaskOnlyAfterSuccessfulWriteback(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: &staticImageExtractor{text: "old caption"},
		},
	})

	if _, err := b.Write("/img/requeue.png", fakePNGAlt, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/img/requeue.png")
	for _, task := range loadSemanticTasksForFile(t, b, fileID) {
		if task.TaskType == string(semantic.TaskTypeEmbed) {
			t.Fatalf("image upload queued embed work before extraction: %+v", task)
		}
	}

	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/requeue.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/requeue.png", 3*time.Second)
	if got != "old caption" {
		t.Fatalf("expected extracted text %q, got %q", "old caption", got)
	}

	tasks := loadSemanticTasksForFile(t, b, fileID)
	var embedQueued int
	for _, task := range tasks {
		if task.TaskType == string(semantic.TaskTypeEmbed) && task.Status == "queued" {
			embedQueued++
		}
	}
	if embedQueued != 1 {
		t.Fatalf("expected 1 queued embed task after image writeback, got %d (all tasks: %+v)", embedQueued, tasks)
	}
}

func TestAsyncImageExtractStaleResultQueuesOnlyCurrentRevision(t *testing.T) {
	extractor := &gatedImageExtractor{
		started: make(chan struct{}, 1),
		release: make(chan struct{}),
	}
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: extractor,
		},
	})

	if _, err := b.Write("/img/stale.png", fakePNGAlt, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/img/stale.png")
	for _, task := range loadSemanticTasksForFile(t, b, fileID) {
		if task.TaskType == string(semantic.TaskTypeEmbed) {
			t.Fatalf("image upload queued embed work before extraction: %+v", task)
		}
	}

	// Start image extraction for revision 1 (gated).
	resultCh := make(chan ImageExtractResult, 1)
	errCh := make(chan error, 1)
	go func() {
		result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
			FileID:      fileID,
			Path:        "/img/stale.png",
			ContentType: "image/png",
			Revision:    1,
		})
		resultCh <- result
		errCh <- err
	}()
	select {
	case <-extractor.started:
	case <-time.After(2 * time.Second):
		t.Fatal("first extraction did not start")
	}

	// Overwrite to revision 2 while revision 1 extraction is in-flight.
	if _, err := b.Write("/img/stale.png", fakePNG, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	close(extractor.release)

	// The stale revision 1 extraction should complete as stale (not write).
	if gotErr := <-errCh; gotErr != nil {
		t.Fatal(gotErr)
	}
	if got := <-resultCh; got != ImageExtractResultStale {
		t.Fatalf("stale result=%q, want %q", got, ImageExtractResultStale)
	}

	// Process revision 2 directly.
	_, revision, _, _ := mustFileForPath(t, b, "/img/stale.png")
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/stale.png",
		ContentType: "image/png",
		Revision:    revision,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/stale.png", 3*time.Second)
	if got != "new caption" {
		t.Fatalf("expected final extracted text %q, got %q", "new caption", got)
	}

	tasks := loadSemanticTasksForFile(t, b, fileID)
	var embedRevisionOne, embedQueued int
	for _, task := range tasks {
		if task.TaskType == string(semantic.TaskTypeEmbed) {
			switch {
			case task.ResourceVersion == 1:
				embedRevisionOne++
			case task.ResourceVersion == 2 && task.Status == "queued":
				embedQueued++
			}
		}
	}
	if embedRevisionOne != 0 {
		t.Fatalf("stale extraction queued %d embed tasks for revision 1 (all tasks: %+v)", embedRevisionOne, tasks)
	}
	if embedQueued != 1 {
		t.Fatalf("expected 1 queued embed task for revision 2, got %d (all tasks: %+v)", embedQueued, tasks)
	}
}

func TestProcessImageExtractTaskWritesContentTextAndQueuesEmbedTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/direct.png", "image/png", fakePNG)
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/direct.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	nf, err := b.Store().Stat(context.Background(), "/img/direct.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/direct.png: %v", err)
	}
	if !strings.Contains(nf.File.ContentText, "cat on sofa") {
		t.Fatalf("unexpected extracted text: %q", nf.File.ContentText)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != "embed" || tasks[0].ResourceVersion != 1 || tasks[0].Status != "queued" {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
}

func TestProcessImageExtractTaskDisabledSemanticWritesTextButSkipsEmbedTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		// AppSemanticTasksEnabled intentionally false — simulates deployment
		// without DRIVE9_EMBED_* worker configured.
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/disabled-embed.png", "image/png", fakePNG)
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/disabled-embed.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	nf, err := b.Store().Stat(context.Background(), "/img/disabled-embed.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/disabled-embed.png: %v", err)
	}
	if !strings.Contains(nf.File.ContentText, "cat on sofa") {
		t.Fatalf("content_text should be written even with disabled semantic tasks, got %q", nf.File.ContentText)
	}
	if tasks := loadSemanticTasksForFile(t, b, fileID); len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0 when AppSemanticTasksEnabled=false", len(tasks))
	}
}

func TestProcessImageExtractTaskStructuredJSONWritesAttributesWithProvenance(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: `{
  "caption_zh": "秋季林荫道路",
  "description_zh": "一条蜿蜒柏油路穿过金黄色树林。",
  "caption_en": "Autumn tree-lined road",
  "description_en": "A winding asphalt road runs through tall trees with golden leaves.",
  "ocr_text": [],
  "tags_zh": ["秋季", "林荫路"],
  "tags_en": ["Autumn", "tree-lined road"],
  "search_queries_zh": ["秋天金黄色树林道路"],
  "search_queries_en": ["autumn golden tree road"],
  "attributes": {"season": "autumn", "scene": "tree-lined road", "style": "landscape"}
}`},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/structured.png", "image/png", fakePNG)
	err := b.Store().InTx(context.Background(), func(tx *sql.Tx) error {
		// User tags: album and drive9.thumbnail.ready must be preserved, scene
		// collides with a model attribute (user wins), legacy numbered
		// drive9.image.* rows must be wiped.
		if err := b.Store().ReplaceFileTagsTx(tx, fileID, map[string]string{
			"album":                  "Inbox",
			"scene":                  "user-scene",
			"drive9.image.tag.en.0":  "old",
			"drive9.thumbnail.ready": "true",
		}); err != nil {
			return err
		}
		// Stale model-owned attributes from a previous extraction: season must
		// be overwritten, mood must be deleted.
		_, err := b.Store().InsertFileTagsWithSourceTx(tx, fileID, datastore.TagSourceImageExtract, map[string]string{
			"season": "winter",
			"mood":   "old-mood",
		})
		return err
	})
	if err != nil {
		t.Fatalf("seed tags: %v", err)
	}

	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/structured.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	nf, err := b.Store().Stat(context.Background(), "/img/structured.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/structured.png: %v", err)
	}
	if strings.Contains(nf.File.ContentText, "caption_zh") || !strings.Contains(nf.File.ContentText, "中文摘要：秋季林荫道路") {
		t.Fatalf("unexpected semantic text: %q", nf.File.ContentText)
	}
	if !strings.Contains(nf.File.ContentText, "属性：scene=tree-lined road，season=autumn，style=landscape") {
		t.Fatalf("attribute values should remain in content_text: %q", nf.File.ContentText)
	}
	tags, err := b.Store().GetFileTags(context.Background(), fileID)
	if err != nil {
		t.Fatalf("GetFileTags: %v", err)
	}
	if tags["album"] != "Inbox" || tags["drive9.thumbnail.ready"] != "true" {
		t.Fatalf("application tags were not preserved: %+v", tags)
	}
	if _, ok := tags["drive9.image.tag.en.0"]; ok {
		t.Fatalf("legacy drive9.image.* rows should be wiped: %+v", tags)
	}
	if tags["scene"] != "user-scene" {
		t.Fatalf("user tag should win over model attribute: %+v", tags)
	}
	if _, ok := tags["mood"]; ok {
		t.Fatalf("stale image_extract attribute should be deleted: %+v", tags)
	}
	if tags["season"] != "autumn" || tags["style"] != "landscape" {
		t.Fatalf("new attributes missing or not replaced: %+v", tags)
	}
	if len(tags) != 5 {
		t.Fatalf("unexpected tag set: %+v", tags)
	}
}

func TestProcessImageExtractTaskFallbackTextClearsModelOwnedTagsOnly(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/fallback-tags.png", "image/png", fakePNG)
	err := b.Store().InTx(context.Background(), func(tx *sql.Tx) error {
		if err := b.Store().ReplaceFileTagsTx(tx, fileID, map[string]string{
			"album":                 "Inbox",
			"drive9.image.schema":   "structured_v1",
			"drive9.image.tag.en.0": "old",
		}); err != nil {
			return err
		}
		// Model-owned attributes from a previous extraction must be wiped even
		// when the new output is non-structured fallback text.
		_, err := b.Store().InsertFileTagsWithSourceTx(tx, fileID, datastore.TagSourceImageExtract, map[string]string{
			"season": "autumn",
		})
		return err
	})
	if err != nil {
		t.Fatalf("seed tags: %v", err)
	}

	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/fallback-tags.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	tags, err := b.Store().GetFileTags(context.Background(), fileID)
	if err != nil {
		t.Fatalf("GetFileTags: %v", err)
	}
	if tags["album"] != "Inbox" || len(tags) != 1 {
		t.Fatalf("tags after fallback writeback = %+v, want only album=Inbox", tags)
	}
}

func TestProcessImageExtractTaskAutoEmbeddingSkipsEmbedBridge(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "cat on sofa screenshot invoice"},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/direct-auto.png", "image/png", fakePNG)
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/direct-auto.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	nf, err := b.Store().Stat(context.Background(), "/img/direct-auto.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/direct-auto.png: %v", err)
	}
	if !strings.Contains(nf.File.ContentText, "cat on sofa") {
		t.Fatalf("unexpected extracted text: %q", nf.File.ContentText)
	}
	if tasks := loadSemanticTasksForFile(t, b, fileID); len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestProcessImageExtractTaskTooLargeReturnsTerminalResult(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:       true,
			Workers:       1,
			QueueSize:     8,
			MaxImageBytes: 4,
			Extractor:     &staticImageExtractor{text: "should not run"},
		},
	})

	fileID := insertImageFileForExtractTest(t, b, "/img/large.png", "image/png", fakePNG)
	result, err := b.ProcessImageExtractTask(context.Background(), ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/large.png",
		ContentType: "image/png",
		Revision:    1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultTooLarge {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultTooLarge)
	}
}

func TestProcessImageExtractTaskWithRuntimeWorksWithoutDefaultExtractor(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			// Capability only: a resolver-backed multi-tenant server with no
			// process-default extractor.
			Capability: true,
		},
	})
	if b.SupportsAsyncImageExtract() {
		t.Fatal("capability-only backend must not report a process-default extract runtime")
	}

	fileID := insertImageFileForExtractTest(t, b, "/img/runtime.png", "image/png", fakePNG)
	task := ImageExtractTaskSpec{
		FileID:      fileID,
		Path:        "/img/runtime.png",
		ContentType: "image/png",
		Revision:    1,
	}

	// The default path refuses: no process-default runtime is configured.
	result, err := b.ProcessImageExtractTask(context.Background(), task)
	if err == nil {
		t.Fatal("ProcessImageExtractTask should fail without a default runtime")
	}
	if result != ImageExtractResultRuntimeNotConfigured {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultRuntimeNotConfigured)
	}

	// The injected-runtime path processes the task standalone.
	result, err = b.ProcessImageExtractTaskWithRuntime(context.Background(), task, ImageExtractRuntime{
		Extractor:           &staticImageExtractor{text: "resolved tenant caption"},
		Timeout:             5 * time.Second,
		MaxImageBytes:       1 << 20,
		MaxExtractTextBytes: 1 << 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	if result != ImageExtractResultWritten {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultWritten)
	}

	got := waitForContentText(t, b, "/img/runtime.png", time.Second)
	if got != "resolved tenant caption" {
		t.Fatalf("unexpected extracted text: %q", got)
	}
}

func TestProcessImageExtractTaskWithRuntimeRejectsNilExtractor(t *testing.T) {
	b := newTestBackend(t)
	result, err := b.ProcessImageExtractTaskWithRuntime(context.Background(),
		ImageExtractTaskSpec{FileID: "file-1", Path: "/img/a.png", Revision: 1},
		ImageExtractRuntime{})
	if err == nil {
		t.Fatal("expected error for nil extractor runtime")
	}
	if result != ImageExtractResultRuntimeNotConfigured {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultRuntimeNotConfigured)
	}
}

func TestProcessImageExtractTaskWithRuntimeRejectsZeroMaxImageBytes(t *testing.T) {
	b := newTestBackend(t)
	result, err := b.ProcessImageExtractTaskWithRuntime(context.Background(),
		ImageExtractTaskSpec{FileID: "file-1", Path: "/img/a.png", Revision: 1},
		ImageExtractRuntime{
			Extractor:     &staticImageExtractor{text: "must not run"},
			MaxImageBytes: 0,
		})
	if err == nil {
		t.Fatal("expected error for zero max image bytes")
	}
	if result != ImageExtractResultRuntimeNotConfigured {
		t.Fatalf("result=%q, want %q", result, ImageExtractResultRuntimeNotConfigured)
	}
}

func insertImageFileForExtractTest(t *testing.T, b *Dat9Backend, path, contentType string, data []byte) string {
	t.Helper()
	fileID := b.genID()
	now := time.Now().UTC()
	digest := sha256.Sum256(data)
	err := b.Store().InTx(context.Background(), func(tx *sql.Tx) error {
		if err := b.Store().InsertFileTx(tx, &datastore.File{
			FileID:         fileID,
			StorageType:    datastore.StorageDB9,
			StorageRef:     "inline",
			ContentBlob:    append([]byte(nil), data...),
			ContentType:    contentType,
			SizeBytes:      int64(len(data)),
			Revision:       1,
			Status:         datastore.StatusConfirmed,
			CreatedAt:      now,
			ConfirmedAt:    &now,
			ChecksumSHA256: hex.EncodeToString(digest[:]),
		}); err != nil {
			return err
		}
		if err := b.Store().EnsureParentDirsTx(tx, path, b.genID); err != nil {
			return err
		}
		return b.Store().InsertNodeTx(tx, &datastore.FileNode{
			NodeID:     b.genID(),
			Path:       path,
			ParentPath: pathutil.ParentPath(path),
			Name:       pathutil.BaseName(path),
			FileID:     fileID,
			CreatedAt:  now,
		})
	})
	if err != nil {
		t.Fatalf("insert image file %s: %v", path, err)
	}
	return fileID
}

func insertS3ImageFileForExtractTest(t *testing.T, b *Dat9Backend, path, contentType string, data []byte) string {
	t.Helper()
	fileID := b.genID()
	storageRef := "image-extract/" + fileID
	ctx := context.Background()
	if err := b.S3().PutObject(ctx, storageRef, bytes.NewReader(data), int64(len(data)), s3client.EncryptionOpts{}); err != nil {
		t.Fatalf("put image object %s: %v", path, err)
	}
	now := time.Now().UTC()
	digest := sha256.Sum256(data)
	err := b.Store().InTx(ctx, func(tx *sql.Tx) error {
		if err := b.Store().InsertFileTx(tx, &datastore.File{
			FileID:         fileID,
			StorageType:    datastore.StorageS3,
			StorageRef:     storageRef,
			ContentType:    contentType,
			SizeBytes:      int64(len(data)),
			Revision:       1,
			Status:         datastore.StatusConfirmed,
			CreatedAt:      now,
			ConfirmedAt:    &now,
			ChecksumSHA256: hex.EncodeToString(digest[:]),
		}); err != nil {
			return err
		}
		if err := b.Store().EnsureParentDirsTx(tx, path, b.genID); err != nil {
			return err
		}
		return b.Store().InsertNodeTx(tx, &datastore.FileNode{
			NodeID:     b.genID(),
			Path:       path,
			ParentPath: pathutil.ParentPath(path),
			Name:       pathutil.BaseName(path),
			FileID:     fileID,
			CreatedAt:  now,
		})
	})
	if err != nil {
		t.Fatalf("insert S3 image file %s: %v", path, err)
	}
	return fileID
}

func waitForContentText(t *testing.T, b *Dat9Backend, path string, timeout time.Duration) string {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		nf, err := b.store.Stat(backgroundWithTrace(), path)
		if err == nil && nf.File != nil {
			if strings.TrimSpace(nf.File.ContentText) != "" {
				return nf.File.ContentText
			}
		}
		time.Sleep(20 * time.Millisecond)
	}
	nf, err := b.store.Stat(backgroundWithTrace(), path)
	if err != nil || nf.File == nil {
		t.Fatalf("file not found while waiting for extracted text: path=%s err=%v", path, err)
	}
	t.Fatalf("timed out waiting for extracted text, last value=%q", nf.File.ContentText)
	return ""
}

// A BeforeExtract guard runs immediately before the provider call: its error
// aborts the task (retryable, surfaced to the caller) without invoking the
// extractor; a nil error lets the extraction through.
func TestProcessImageExtractTaskWithRuntimeBeforeExtractGuard(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{Capability: true},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/guard.png", "image/png", fakePNG)
	task := ImageExtractTaskSpec{FileID: fileID, Path: "/img/guard.png", ContentType: "image/png", Revision: 1}

	guardErr := errors.New("image extract globally disabled")
	spy := &countingImageExtractor{text: "caption"}
	runtime := ImageExtractRuntime{
		Extractor:           spy,
		Timeout:             5 * time.Second,
		MaxImageBytes:       1 << 20,
		MaxExtractTextBytes: 1 << 10,
		BeforeExtract:       func() error { return guardErr },
	}
	result, err := b.ProcessImageExtractTaskWithRuntime(context.Background(), task, runtime)
	if !errors.Is(err, guardErr) {
		t.Fatalf("err = %v, want the guard error", err)
	}
	if result == ImageExtractResultWritten {
		t.Fatalf("result = %q, want a non-written outcome", result)
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0 (provider must not be called)", spy.calls)
	}

	runtime.BeforeExtract = func() error { return nil }
	result, err = b.ProcessImageExtractTaskWithRuntime(context.Background(), task, runtime)
	if err != nil || result != ImageExtractResultWritten {
		t.Fatalf("result = %q err = %v, want written", result, err)
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spy.calls)
	}
}

// The process-default path accepts the same guard: a failing guard aborts
// before the provider call; a nil guard keeps ProcessImageExtractTask
// behavior.
func TestProcessImageExtractTaskWithGuard(t *testing.T) {
	spy := &countingImageExtractor{text: "default caption"}
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{Enabled: true, Extractor: spy},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/guard-default.png", "image/png", fakePNG)
	task := ImageExtractTaskSpec{FileID: fileID, Path: "/img/guard-default.png", ContentType: "image/png", Revision: 1}

	guardErr := errors.New("image extract globally disabled")
	result, err := b.ProcessImageExtractTaskWithGuard(context.Background(), task, func() error { return guardErr })
	if !errors.Is(err, guardErr) {
		t.Fatalf("err = %v, want the guard error", err)
	}
	if result == ImageExtractResultWritten {
		t.Fatalf("result = %q, want a non-written outcome", result)
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0 (provider must not be called)", spy.calls)
	}

	result, err = b.ProcessImageExtractTaskWithGuard(context.Background(), task, nil)
	if err != nil || result != ImageExtractResultWritten {
		t.Fatalf("result = %q err = %v, want written", result, err)
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spy.calls)
	}
}