package server

import (
	"context"
	"database/sql"
	"encoding/json"
	"strings"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/tenant"
	tenantschema "github.com/mem9-ai/drive9/pkg/tenant/schema"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

// installSharedCoreFSSchemaForExtract swaps the Core FS tables (and drops the
// standalone llm_usage) to the shared (fs_id) shape for the duration of the
// test. The cleanup restores the standalone schema. Package tests that touch
// the shared test database run sequentially, so the swap is safe (same
// pattern as pkg/datastore's installSharedCoreFSSchema and this package's
// installSharedFSEventsTable).
func installSharedCoreFSSchemaForExtract(t *testing.T) {
	t.Helper()
	tables := []string{
		"file_nodes", "inodes", "contents", "semantic", "file_tags",
		"uploads", "semantic_tasks", "file_gc_tasks", "fs_events", "llm_usage",
	}
	drop := func(db *sql.DB) {
		t.Helper()
		for _, tbl := range tables {
			if _, err := db.Exec("DROP TABLE IF EXISTS " + tbl); err != nil {
				t.Fatalf("drop %s: %v", tbl, err)
			}
		}
	}
	db, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = db.Close() }()
	drop(db)
	t.Cleanup(func() {
		db, err := sql.Open("mysql", testDSN)
		if err != nil {
			t.Errorf("reopen test db for schema restore: %v", err)
			return
		}
		defer func() { _ = db.Close() }()
		drop(db)
		initServerTenantSchema(t, testDSN)
	})
	for _, stmt := range tenantschema.CoreFSMySQLSharedSchemaStatements() {
		if _, err := db.Exec(stmt); err != nil {
			t.Fatalf("apply shared schema: %v", err)
		}
	}
}

// newSharedExtractBackend builds a backend bound to the shared schema with
// the given fs_id, exactly the shape pool.createBackend produces for shared
// tenants: capability-only image extract, no app semantic tasks, no database
// auto-embedding, no audio extraction.
func newSharedExtractBackend(t *testing.T, fsID int64, opts backend.Options) *backend.Dat9Backend {
	t.Helper()
	store, err := datastore.OpenForTenantScoped(context.Background(), testDSN, "", "", datastore.SharedScope(fsID))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, opts)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(b.Close)
	return b
}

func sharedExtractCapabilityOptions() backend.Options {
	return backend.Options{AsyncImageExtract: backend.AsyncImageExtractOptions{Capability: true}}
}

func sharedVideoExtractCapabilityOptions(maxVideoFiles int64) backend.Options {
	return backend.Options{AsyncVideoExtract: backend.AsyncVideoExtractOptions{
		Capability:       true,
		MaxVideoLLMFiles: maxVideoFiles,
	}}
}

func writeSharedImage(t *testing.T, b *backend.Dat9Backend, path string) {
	t.Helper()
	if _, err := b.Write(path, fakeImagePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatalf("write %s: %v", path, err)
	}
}

func sharedExtractTarget(t *testing.T, tenantID string, b *backend.Dat9Backend) *tenantTarget {
	t.Helper()
	return &tenantTarget{
		tenantID:         tenantID,
		backend:          b,
		store:            b.Store(),
		allowedTaskTypes: []semantic.TaskType{semantic.TaskTypeImgExtractText},
		release:          func() {},
	}
}

func newSharedExtractWorker(b *backend.Dat9Backend, resolver *extractConfigResolver) *tenantWorkerManager {
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	if m == nil {
		return nil
	}
	m.setExtractRuntimeDeps(resolver, nil)
	return m
}

// countFileTags counts attribute rows for one file inside one tenant scope.
// The raw query must filter fs_id explicitly: store.DB() bypasses the scope
// rewriting that the Store's own methods apply.
func countFileTags(t *testing.T, store *datastore.Store, fsID int64, fileID string) int {
	t.Helper()
	var n int
	if err := store.DB().QueryRowContext(context.Background(),
		`SELECT COUNT(*) FROM file_tags WHERE fs_id = ? AND file_id = ?`, fsID, fileID).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func contentTextOf(t *testing.T, b *backend.Dat9Backend, path string) string {
	t.Helper()
	nf, err := b.Store().Stat(context.Background(), path)
	if err != nil || nf.File == nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	return nf.File.ContentText
}

// A shared tenant with a custom config executes img_extract_text even when
// the process-default extractor is off (resolver def is nil). With
// AppSemanticTasksEnabled=false the writeback must not enqueue an embed task.
func TestSharedTenantImgExtractCustomConfigWithoutDefault(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	b := newSharedExtractBackend(t, 8800001, sharedExtractCapabilityOptions())
	writeSharedImage(t, b, "/img/shared-custom.png")

	spy := &spyImageExtractor{text: `{"caption_en":"shared custom caption","attributes":{"scene":"office"}}`}
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t-shared-custom": enabledImageRow("t-shared-custom"),
	}}
	resolver := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "k"}, nil, time.Minute)
	t.Cleanup(resolver.Close)
	resolver.newOpenAIExtractor = func(backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		return spy, nil
	}
	m := newSharedExtractWorker(b, resolver)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	target := sharedExtractTarget(t, "t-shared-custom", b)

	if claimed, _ := m.claimAndProcessOne(context.Background(), target); !claimed {
		t.Fatal("no img_extract_text claim for shared tenant")
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spy.calls)
	}
	if got := contentTextOf(t, b, "/img/shared-custom.png"); !strings.Contains(got, "shared custom caption") {
		t.Fatalf("content_text = %q, want structured caption", got)
	}
	if got := countFileTags(t, b.Store(), 8800001, fileIDOf(t, b, "/img/shared-custom.png")); got == 0 {
		t.Fatal("attribute tags were not written for shared tenant")
	}
	assertSemanticTaskCount(t, b.Store(), string(semantic.TaskQueued), 0)

	// AppSemanticTasksEnabled=false on shared tenants: the image writeback
	// must not have enqueued an embed task at any point.
	var embedRows int
	if err := b.Store().DB().QueryRowContext(context.Background(),
		`SELECT COUNT(*) FROM semantic_tasks WHERE fs_id = ? AND task_type = ?`, int64(8800001), string(semantic.TaskTypeEmbed)).Scan(&embedRows); err != nil {
		t.Fatal(err)
	}
	if embedRows != 0 {
		t.Fatalf("embed task rows = %d, want 0 for shared tenant", embedRows)
	}
}

func fileIDOf(t *testing.T, b *backend.Dat9Backend, path string) string {
	t.Helper()
	nf, err := b.Store().Stat(context.Background(), path)
	if err != nil || nf.File == nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	return nf.File.FileID
}

// A shared tenant with no config row runs the process-default extractor when
// one is configured.
func TestSharedTenantImgExtractAbsentRowUsesDefault(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	b := newSharedExtractBackend(t, 8800002, sharedExtractCapabilityOptions())
	writeSharedImage(t, b, "/img/shared-default.png")

	spy := &spyImageExtractor{text: "default runtime caption"}
	def := defaultTestRuntime()
	def.Extractor = spy
	resolver := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, def, time.Minute)
	t.Cleanup(resolver.Close)
	m := newSharedExtractWorker(b, resolver)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	target := sharedExtractTarget(t, "t-shared-default", b)

	if claimed, _ := m.claimAndProcessOne(context.Background(), target); !claimed {
		t.Fatal("no img_extract_text claim for shared tenant")
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1 (default runtime)", spy.calls)
	}
	if got := contentTextOf(t, b, "/img/shared-default.png"); got != "default runtime caption" {
		t.Fatalf("content_text = %q, want default runtime caption", got)
	}
	assertSemanticTaskCount(t, b.Store(), string(semantic.TaskQueued), 0)
}

// Shared tenants get terminal discards for the off-marker (disabled) and for
// absent-row-without-default (not_configured); the backend is never touched.
func TestSharedTenantImgExtractTerminalStates(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	b := newSharedExtractBackend(t, 8800003, sharedExtractCapabilityOptions())

	t.Run("off marker discards disabled", func(t *testing.T) {
		reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
			"t-shared-off": disabledImageRow("t-shared-off"),
		}}
		resolver := newExtractConfigResolver(reader, &fakeExtractDecryptor{}, defaultTestRuntime(), time.Minute)
		t.Cleanup(resolver.Close)
		m := &tenantWorkerManager{extractResolver: resolver}
		outcome := m.processImgExtractTask(context.Background(), &tenantTarget{tenantID: "t-shared-off", backend: b}, imgExtractTestTask())
		if outcome.action != semanticTaskActionDiscard || outcome.result != "disabled" {
			t.Fatalf("outcome = %+v, want discard/disabled", outcome)
		}
	})

	t.Run("absent row without default discards not_configured", func(t *testing.T) {
		resolver := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
			&fakeExtractDecryptor{}, nil, time.Minute)
		t.Cleanup(resolver.Close)
		m := &tenantWorkerManager{extractResolver: resolver}
		outcome := m.processImgExtractTask(context.Background(), &tenantTarget{tenantID: "t-shared-none", backend: b}, imgExtractTestTask())
		if outcome.action != semanticTaskActionDiscard || outcome.result != "not_configured" {
			t.Fatalf("outcome = %+v, want discard/not_configured", outcome)
		}
	})
}

// Two shared tenants whose tasks reference the same file_id string stay
// isolated by fs_id: tasks, content_text, and attributes never cross scopes.
func TestSharedTenantImgExtractFsIDIsolation(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	ctx := context.Background()
	backendA := newSharedExtractBackend(t, 8800011, sharedExtractCapabilityOptions())
	backendB := newSharedExtractBackend(t, 8800012, sharedExtractCapabilityOptions())

	writeSharedImage(t, backendA, "/img/dup.png")
	fileIDA := fileIDOf(t, backendA, "/img/dup.png")

	// Tenant B's queue carries a task for the SAME file_id string (in B's
	// scope that file does not exist).
	payload, err := json.Marshal(semantic.ImgExtractTaskPayload{Path: "/img/dup.png", ContentType: "image/png"})
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	if _, err := backendB.Store().EnqueueSemanticTask(ctx, &semantic.Task{
		TaskID: "task-b-shared-fileid", TaskType: semantic.TaskTypeImgExtractText,
		ResourceID: fileIDA, ResourceVersion: 1, Status: semantic.TaskQueued,
		MaxAttempts: 5, AvailableAt: now, CreatedAt: now, UpdatedAt: now, PayloadJSON: payload,
	}); err != nil {
		t.Fatal(err)
	}

	spyA := &spyImageExtractor{text: `{"caption_en":"tenant A caption","attributes":{"owner":"a"}}`}
	resolver := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, &backend.ImageExtractRuntime{
			Extractor: spyA, Timeout: 5 * time.Second, MaxImageBytes: 8 << 20,
			MaxExtractTextBytes: backend.DefaultImageExtractMaxTextBytes,
		}, time.Minute)
	t.Cleanup(resolver.Close)
	m := newSharedExtractWorker(backendA, resolver)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	targetA := sharedExtractTarget(t, "tenant-a", backendA)
	targetB := sharedExtractTarget(t, "tenant-b", backendB)

	// Drain A: extraction writes content_text + attributes in A's scope only.
	if claimed, _ := m.claimAndProcessOne(ctx, targetA); !claimed {
		t.Fatal("no img_extract_text claim for tenant A")
	}
	if spyA.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spyA.calls)
	}
	if got := contentTextOf(t, backendA, "/img/dup.png"); !strings.Contains(got, "tenant A caption") {
		t.Fatalf("tenant A content_text = %q", got)
	}
	if got := countFileTags(t, backendA.Store(), 8800011, fileIDA); got == 0 {
		t.Fatal("tenant A attributes were not written")
	}

	// Drain B: B's task for the same file_id string resolves inside B's
	// scope, where the file does not exist → terminal file_not_found ACK.
	if claimed, _ := m.claimAndProcessOne(ctx, targetB); !claimed {
		t.Fatal("no img_extract_text claim for tenant B")
	}
	assertSemanticTaskCount(t, backendB.Store(), string(semantic.TaskQueued), 0)

	// Isolation: B has no file, no content_text, and no attributes for the
	// shared file_id string; A's state is untouched by B's drain.
	if _, err := backendB.Store().Stat(ctx, "/img/dup.png"); err == nil {
		t.Fatal("tenant B must not see tenant A's file")
	}
	if got := countFileTags(t, backendB.Store(), 8800012, fileIDA); got != 0 {
		t.Fatalf("tenant B file_tags for shared file_id = %d, want 0", got)
	}
	if got := contentTextOf(t, backendA, "/img/dup.png"); !strings.Contains(got, "tenant A caption") {
		t.Fatalf("tenant A content_text changed after B drain: %q", got)
	}
	if got := countFileTags(t, backendA.Store(), 8800011, fileIDA); got == 0 {
		t.Fatal("tenant A attributes lost after B drain")
	}
	if spyA.calls != 1 {
		t.Fatalf("extractor calls after B drain = %d, want 1", spyA.calls)
	}
}

func TestSharedTenantVideoQuotaIsScopedByFSID(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	ctx := context.Background()
	backendA := newSharedExtractBackend(t, 8800031, sharedVideoExtractCapabilityOptions(1))
	backendB := newSharedExtractBackend(t, 8800032, sharedVideoExtractCapabilityOptions(1))

	if _, err := backendB.Write("/b.mp4", []byte("tenant-b-video"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	if _, err := backendA.Write("/a.mp4", []byte("tenant-a-video"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	for _, tc := range []struct {
		name  string
		store *datastore.Store
		fsID  int64
	}{
		{name: "tenant A", store: backendA.Store(), fsID: 8800031},
		{name: "tenant B", store: backendB.Store(), fsID: 8800032},
	} {
		var count int
		if err := tc.store.DB().QueryRowContext(ctx,
			`SELECT COUNT(*) FROM semantic_tasks WHERE fs_id = ? AND task_type = ?`,
			tc.fsID, string(semantic.TaskTypeVideoExtractVisual)).Scan(&count); err != nil {
			t.Fatal(err)
		}
		if count != 1 {
			t.Fatalf("%s video tasks = %d, want 1; another fs_id must not consume its quota", tc.name, count)
		}
	}
}

// seedSharedQuotaTenant inserts the meta rows the central quota path needs
// and returns the tenant ID.
func seedSharedQuotaTenant(t *testing.T, metaStore *meta.Store, tenantID string, cfg *meta.QuotaConfig) {
	t.Helper()
	ctx := context.Background()
	now := time.Now().UTC()
	if err := metaStore.InsertTenant(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantActive, Kind: meta.TenantKindLive,
		Provider: tenant.ProviderTiDBCloudNativeShared, DBPasswordCipher: []byte{},
		SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	cfg.TenantID = tenantID
	if err := metaStore.SetQuotaConfig(ctx, cfg); err != nil {
		t.Fatal(err)
	}
	if err := metaStore.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
}

func openSharedQuotaMeta(t *testing.T) *meta.Store {
	t.Helper()
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	return metaStore
}

// The media-LLM file quota still gates task enqueue on the shared path: an
// over-quota image write succeeds but enqueues no img_extract_text task.
func TestSharedTenantImgExtractMediaQuotaSkipsEnqueue(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	metaStore := openSharedQuotaMeta(t)
	ctx := context.Background()
	const tenantID = "t-shared-media-quota"
	seedSharedQuotaTenant(t, metaStore, tenantID, &meta.QuotaConfig{
		MaxStorageBytes:  1 << 30,
		MaxFileSizeBytes: 1 << 30,
		MaxMediaLLMFiles: 1,
		MaxMonthlyCostMC: 1 << 30,
	})
	if err := metaStore.IncrMediaFileCount(ctx, tenantID, 1); err != nil {
		t.Fatal(err)
	}

	b := newSharedExtractBackend(t, 8800021, sharedExtractCapabilityOptions())
	b.SetMetaQuotaStore(ctx, tenantID, tenant.NewMetaQuotaAdapter(metaStore))

	writeSharedImage(t, b, "/img/over-quota.png")
	var imgTasks int
	if err := b.Store().DB().QueryRowContext(ctx,
		`SELECT COUNT(*) FROM semantic_tasks WHERE fs_id = ? AND task_type = ?`, int64(8800021), string(semantic.TaskTypeImgExtractText)).Scan(&imgTasks); err != nil {
		t.Fatal(err)
	}
	if imgTasks != 0 {
		t.Fatalf("img_extract_text tasks = %d, want 0 when the media LLM quota is exceeded", imgTasks)
	}
}

// The monthly LLM cost budget still gates execution on the shared path: an
// exhausted budget ACKs the task budget_exhausted without calling the
// provider.
func TestSharedTenantImgExtractMonthlyBudgetGate(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	metaStore := openSharedQuotaMeta(t)
	ctx := context.Background()
	const tenantID = "t-shared-budget"
	seedSharedQuotaTenant(t, metaStore, tenantID, &meta.QuotaConfig{
		MaxStorageBytes:  1 << 30,
		MaxFileSizeBytes: 1 << 30,
		MaxMediaLLMFiles: 100,
		MaxMonthlyCostMC: 100,
	})
	if err := metaStore.IncrMonthlyLLMCost(ctx, tenantID, 200); err != nil {
		t.Fatal(err)
	}

	b := newSharedExtractBackend(t, 8800022, sharedExtractCapabilityOptions())
	b.SetMetaQuotaStore(ctx, tenantID, tenant.NewMetaQuotaAdapter(metaStore))
	writeSharedImage(t, b, "/img/budget.png")

	spy := &spyImageExtractor{text: "never called"}
	resolver := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, &backend.ImageExtractRuntime{
			Extractor: spy, Timeout: 5 * time.Second, MaxImageBytes: 8 << 20,
			MaxExtractTextBytes: backend.DefaultImageExtractMaxTextBytes,
		}, time.Minute)
	t.Cleanup(resolver.Close)
	m := newSharedExtractWorker(b, resolver)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	target := sharedExtractTarget(t, tenantID, b)

	if claimed, _ := m.claimAndProcessOne(ctx, target); !claimed {
		t.Fatal("no img_extract_text claim for shared tenant")
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0 when the monthly budget is exhausted", spy.calls)
	}
	assertSemanticTaskCount(t, b.Store(), string(semantic.TaskQueued), 0)
	if got := contentTextOf(t, b, "/img/budget.png"); got != "" {
		t.Fatalf("content_text = %q, want empty when the budget gate fires", got)
	}
}

// The usage ledger still records extraction cost on the shared path: the
// central tenant_llm_usage row and monthly counter are written even though
// the shared schema has no tenant-local llm_usage table.
func TestSharedTenantImgExtractUsageLedger(t *testing.T) {
	installSharedCoreFSSchemaForExtract(t)
	metaStore := openSharedQuotaMeta(t)
	ctx := context.Background()
	const tenantID = "t-shared-ledger"
	seedSharedQuotaTenant(t, metaStore, tenantID, &meta.QuotaConfig{
		MaxStorageBytes:  1 << 30,
		MaxFileSizeBytes: 1 << 30,
		MaxMediaLLMFiles: 100,
		MaxMonthlyCostMC: 1 << 30,
	})

	opts := sharedExtractCapabilityOptions()
	opts.LLMCostBudget = backend.LLMCostBudgetOptions{VisionCostPerKTokenMillicents: 1000}
	b := newSharedExtractBackend(t, 8800023, opts)
	b.SetMetaQuotaStore(ctx, tenantID, tenant.NewMetaQuotaAdapter(metaStore))
	writeSharedImage(t, b, "/img/ledger.png")
	fileID := fileIDOf(t, b, "/img/ledger.png")

	spy := &spyImageExtractor{text: "ledger caption"}
	spy.usage = backend.ImageExtractUsage{PromptTokens: 120, CompletionTokens: 80}
	resolver := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, &backend.ImageExtractRuntime{
			Extractor: spy, Timeout: 5 * time.Second, MaxImageBytes: 8 << 20,
			MaxExtractTextBytes: backend.DefaultImageExtractMaxTextBytes,
		}, time.Minute)
	t.Cleanup(resolver.Close)
	m := newSharedExtractWorker(b, resolver)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	if claimed, _ := m.claimAndProcessOne(ctx, sharedExtractTarget(t, tenantID, b)); !claimed {
		t.Fatal("no img_extract_text claim for shared tenant")
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spy.calls)
	}

	var ledgerRows int
	if err := metaStore.DB().QueryRowContext(ctx,
		`SELECT COUNT(*) FROM tenant_llm_usage WHERE tenant_id = ? AND task_type = 'img_extract_text' AND task_id = ?`,
		tenantID, fileID).Scan(&ledgerRows); err != nil {
		t.Fatal(err)
	}
	if ledgerRows != 1 {
		t.Fatalf("central llm usage rows = %d, want 1", ledgerRows)
	}
	total, err := metaStore.MonthlyLLMCostMillicents(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	// 200 tokens at 1000 millicents per 1K tokens → 200 millicents.
	if total != 200 {
		t.Fatalf("monthly llm cost = %d, want 200", total)
	}
}