Improve large library storage backends and scans

internal/repository/opensearch.go

@@ -0,0 +1,212 @@
+package repository
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/ShukeBta/MediaStationGo/internal/config"
+	"github.com/ShukeBta/MediaStationGo/internal/model"
+)
+
+type OpenSearchMediaBackend struct {
+	baseURL  string
+	index    string
+	username string
+	password string
+	client   *http.Client
+}
+
+func NewOpenSearchMediaBackend(cfg config.SearchConfig) *OpenSearchMediaBackend {
+	if strings.TrimSpace(cfg.Backend) != "opensearch" || strings.TrimSpace(cfg.OpenSearchURL) == "" {
+		return nil
+	}
+	index := strings.TrimSpace(cfg.Index)
+	if index == "" {
+		index = "mediastation_media"
+	}
+	return &OpenSearchMediaBackend{
+		baseURL:  strings.TrimRight(strings.TrimSpace(cfg.OpenSearchURL), "/"),
+		index:    index,
+		username: strings.TrimSpace(cfg.Username),
+		password: cfg.Password,
+		client:   &http.Client{Timeout: 4 * time.Second},
+	}
+}
+
+func (b *OpenSearchMediaBackend) SearchMediaIDs(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]string, int64, error) {
+	if b == nil || b.client == nil || b.baseURL == "" || b.index == "" {
+		return nil, 0, fmt.Errorf("opensearch backend not configured")
+	}
+	if limit <= 0 {
+		limit = 50
+	}
+	must := []any{
+		map[string]any{
+			"multi_match": map[string]any{
+				"query":     query,
+				"fields":    []string{"title^4", "original_name^3", "genres^2", "path"},
+				"type":      "best_fields",
+				"operator":  "and",
+				"fuzziness": "AUTO",
+			},
+		},
+	}
+	filters := []any{
+		map[string]any{"term": map[string]any{"deleted": false}},
+	}
+	if !filter.IncludeNSFW {
+		filters = append(filters, map[string]any{"term": map[string]any{"nsfw": false}})
+	}
+	if len(filter.AllowedLibraryIDs) > 0 {
+		filters = append(filters, map[string]any{"terms": map[string]any{"library_id": filter.AllowedLibraryIDs}})
+	}
+	if len(filter.HiddenLibraryIDs) > 0 {
+		filters = append(filters, map[string]any{"bool": map[string]any{
+			"must_not": []any{map[string]any{"terms": map[string]any{"library_id": filter.HiddenLibraryIDs}}},
+		}})
+	}
+	body := map[string]any{
+		"from": offset,
+		"size": limit,
+		"_source": []string{
+			"id",
+		},
+		"query": map[string]any{
+			"bool": map[string]any{
+				"must":   must,
+				"filter": filters,
+			},
+		},
+	}
+	var resp struct {
+		Hits struct {
+			Total any `json:"total"`
+			Hits  []struct {
+				ID     string `json:"_id"`
+				Source struct {
+					ID string `json:"id"`
+				} `json:"_source"`
+			} `json:"hits"`
+		} `json:"hits"`
+	}
+	if err := b.doJSON(ctx, http.MethodPost, "/"+url.PathEscape(b.index)+"/_search", body, &resp); err != nil {
+		return nil, 0, err
+	}
+	ids := make([]string, 0, len(resp.Hits.Hits))
+	for _, hit := range resp.Hits.Hits {
+		id := strings.TrimSpace(hit.Source.ID)
+		if id == "" {
+			id = strings.TrimSpace(hit.ID)
+		}
+		if id != "" {
+			ids = append(ids, id)
+		}
+	}
+	return ids, openSearchTotal(resp.Hits.Total), nil
+}
+
+func (b *OpenSearchMediaBackend) EnsureIndex(ctx context.Context) error {
+	if err := b.do(ctx, http.MethodHead, "/"+url.PathEscape(b.index), nil, "", nil); err == nil {
+		return nil
+	}
+	mapping := map[string]any{
+		"mappings": map[string]any{
+			"properties": map[string]any{
+				"id":            map[string]any{"type": "keyword"},
+				"library_id":    map[string]any{"type": "keyword"},
+				"title":         map[string]any{"type": "text"},
+				"original_name": map[string]any{"type": "text"},
+				"path":          map[string]any{"type": "text"},
+				"genres":        map[string]any{"type": "text"},
+				"nsfw":          map[string]any{"type": "boolean"},
+				"deleted":       map[string]any{"type": "boolean"},
+				"created_at":    map[string]any{"type": "date"},
+			},
+		},
+	}
+	return b.doJSON(ctx, http.MethodPut, "/"+url.PathEscape(b.index), mapping, nil)
+}
+
+func (b *OpenSearchMediaBackend) IndexMedia(ctx context.Context, rows []model.Media) error {
+	if len(rows) == 0 {
+		return nil
+	}
+	var bulk bytes.Buffer
+	enc := json.NewEncoder(&bulk)
+	for _, row := range rows {
+		if err := enc.Encode(map[string]any{"index": map[string]any{"_index": b.index, "_id": row.ID}}); err != nil {
+			return err
+		}
+		if err := enc.Encode(map[string]any{
+			"id":            row.ID,
+			"library_id":    row.LibraryID,
+			"title":         row.Title,
+			"original_name": row.OriginalName,
+			"path":          row.Path,
+			"genres":        row.Genres,
+			"nsfw":          row.NSFW,
+			"deleted":       row.DeletedAt.Valid,
+			"created_at":    row.CreatedAt,
+		}); err != nil {
+			return err
+		}
+	}
+	return b.do(ctx, http.MethodPost, "/_bulk", &bulk, "application/x-ndjson", nil)
+}
+
+func (b *OpenSearchMediaBackend) doJSON(ctx context.Context, method, path string, body any, out any) error {
+	var reader io.Reader
+	if body != nil {
+		raw, err := json.Marshal(body)
+		if err != nil {
+			return err
+		}
+		reader = bytes.NewReader(raw)
+	}
+	return b.do(ctx, method, path, reader, "application/json", out)
+}
+
+func (b *OpenSearchMediaBackend) do(ctx context.Context, method, path string, body io.Reader, contentType string, out any) error {
+	req, err := http.NewRequestWithContext(ctx, method, b.baseURL+path, body)
+	if err != nil {
+		return err
+	}
+	if contentType != "" {
+		req.Header.Set("Content-Type", contentType)
+	}
+	if b.username != "" {
+		req.SetBasicAuth(b.username, b.password)
+	}
+	resp, err := b.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		return fmt.Errorf("opensearch %s %s returned %d", method, path, resp.StatusCode)
+	}
+	if out == nil {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return nil
+	}
+	return json.NewDecoder(resp.Body).Decode(out)
+}
+
+func openSearchTotal(value any) int64 {
+	switch v := value.(type) {
+	case float64:
+		return int64(v)
+	case map[string]any:
+		if n, ok := v["value"].(float64); ok {
+			return int64(n)
+		}
+	}
+	return 0
+}

internal/repository/opensearch_test.go

@@ -0,0 +1,57 @@
+package repository
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/ShukeBta/MediaStationGo/internal/config"
+)
+
+func TestOpenSearchMediaBackendSearchesIDs(t *testing.T) {
+	var gotPath string
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		if r.Method != http.MethodPost {
+			t.Fatalf("method = %s, want POST", r.Method)
+		}
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatal(err)
+		}
+		if body["from"].(float64) != 5 || body["size"].(float64) != 10 {
+			t.Fatalf("paging body = %#v", body)
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"hits": map[string]any{
+				"total": map[string]any{"value": 2},
+				"hits": []any{
+					map[string]any{"_id": "m-1", "_source": map[string]any{"id": "m-1"}},
+					map[string]any{"_id": "m-2", "_source": map[string]any{"id": "m-2"}},
+				},
+			},
+		})
+	}))
+	defer upstream.Close()
+
+	backend := NewOpenSearchMediaBackend(config.SearchConfig{
+		Backend:       "opensearch",
+		OpenSearchURL: upstream.URL,
+		Index:         "media-test",
+	})
+	ids, total, err := backend.SearchMediaIDs(t.Context(), "流浪地球", 5, 10, MediaQueryFilter{
+		IncludeNSFW:       false,
+		AllowedLibraryIDs: []string{"lib-1"},
+		HiddenLibraryIDs:  []string{"adult"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotPath != "/media-test/_search" {
+		t.Fatalf("path = %q", gotPath)
+	}
+	if total != 2 || len(ids) != 2 || ids[0] != "m-1" || ids[1] != "m-2" {
+		t.Fatalf("ids=%#v total=%d", ids, total)
+	}
+}

internal/repository/repository.go

@@ -274,6 +274,23 @@ type MediaRepository struct {
 
 	searchIndexOnce      sync.Once
 	searchIndexAvailable bool
+	searchBackend        MediaSearchBackend
+}
+
+type MediaSearchBackend interface {
+	SearchMediaIDs(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]string, int64, error)
+}
+
+type MediaSearchSyncBackend interface {
+	MediaSearchBackend
+	EnsureIndex(ctx context.Context) error
+	IndexMedia(ctx context.Context, rows []model.Media) error
+}
+
+func (r *MediaRepository) SetSearchBackend(backend MediaSearchBackend) {
+	if r != nil {
+		r.searchBackend = backend
+	}
 }
 
 // MediaQueryFilter is applied to user-facing media queries so NSFW items and
@@ -323,6 +340,7 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error {
 			m.ScrapeStatus = "pending"
 		}
 		if createErr := r.db.WithContext(ctx).Create(m).Error; createErr == nil {
+			r.indexMediaBestEffort(ctx, *m)
 			return nil
 		} else if retryErr := r.db.WithContext(ctx).Unscoped().Where("path = ?", m.Path).First(&existing).Error; retryErr != nil {
 			return createErr
@@ -433,9 +451,20 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error {
 	}
 	// 回写 ID / 不可变字段，让 caller 拿到完整的现有行。
 	*m = existing
+	if fresh, err := r.FindByID(ctx, existing.ID); err == nil && fresh != nil {
+		r.indexMediaBestEffort(ctx, *fresh)
+	}
 	return nil
 }
 
+func (r *MediaRepository) indexMediaBestEffort(ctx context.Context, media model.Media) {
+	backend, ok := r.searchBackend.(MediaSearchSyncBackend)
+	if !ok {
+		return
+	}
+	_ = backend.IndexMedia(ctx, []model.Media{media})
+}
+
 func setIfChanged[T comparable](updates map[string]any, key string, current, next T) {
 	if current != next {
 		updates[key] = next
@@ -500,6 +529,11 @@ func (r *MediaRepository) SearchFilteredPage(ctx context.Context, query string,
 	if limit <= 0 {
 		limit = 50
 	}
+	if query != "" && r.searchBackend != nil {
+		if items, total, ok := r.searchFilteredBackend(ctx, query, offset, limit, filter); ok {
+			return items, total, nil
+		}
+	}
 	if query != "" {
 		if items, total, ok := r.searchFilteredFTS(ctx, query, offset, limit, filter); ok {
 			if total > 0 {
@@ -510,6 +544,36 @@ func (r *MediaRepository) SearchFilteredPage(ctx context.Context, query string,
 	return r.searchFilteredLIKE(ctx, query, offset, limit, filter)
 }
 
+func (r *MediaRepository) searchFilteredBackend(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]model.Media, int64, bool) {
+	ids, total, err := r.searchBackend.SearchMediaIDs(ctx, query, offset, limit, filter)
+	if err != nil {
+		return nil, 0, false
+	}
+	if len(ids) == 0 {
+		return []model.Media{}, total, true
+	}
+	var rows []model.Media
+	q := r.db.WithContext(ctx).Model(&model.Media{}).Where("id IN ?", ids)
+	q = applyMediaQueryFilter(q, filter)
+	if err := q.Find(&rows).Error; err != nil {
+		return nil, 0, false
+	}
+	byID := make(map[string]model.Media, len(rows))
+	for _, row := range rows {
+		byID[row.ID] = row
+	}
+	items := make([]model.Media, 0, len(ids))
+	for _, id := range ids {
+		if row, ok := byID[id]; ok {
+			items = append(items, row)
+		}
+	}
+	if len(items) == 0 && total > 0 {
+		return nil, 0, false
+	}
+	return items, total, true
+}
+
 func (r *MediaRepository) searchFilteredFTS(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]model.Media, int64, bool) {
 	if !r.searchIndexEnabled(ctx) {
 		return nil, 0, false
@@ -630,6 +694,9 @@ func escapeLike(value string) string {
 }
 
 func (r *MediaRepository) BackfillSearchIndex(ctx context.Context, batchLimit int) (int64, error) {
+	if backend, ok := r.searchBackend.(MediaSearchSyncBackend); ok {
+		return r.backfillExternalSearchIndex(ctx, backend, batchLimit)
+	}
 	if batchLimit <= 0 {
 		batchLimit = 1000
 	}
@@ -654,10 +721,45 @@ LIMIT ?
 	return res.RowsAffected, res.Error
 }
 
+func (r *MediaRepository) backfillExternalSearchIndex(ctx context.Context, backend MediaSearchSyncBackend, batchLimit int) (int64, error) {
+	if batchLimit <= 0 {
+		batchLimit = 1000
+	}
+	if err := backend.EnsureIndex(ctx); err != nil {
+		return 0, err
+	}
+	var lastID string
+	for {
+		var rows []model.Media
+		q := r.db.WithContext(ctx).
+			Model(&model.Media{}).
+			Where("deleted_at IS NULL")
+		if lastID != "" {
+			q = q.Where("id > ?", lastID)
+		}
+		if err := q.Order("id ASC").Limit(batchLimit).Find(&rows).Error; err != nil {
+			return 0, err
+		}
+		if len(rows) == 0 {
+			return 0, nil
+		}
+		if err := backend.IndexMedia(ctx, rows); err != nil {
+			return 0, err
+		}
+		lastID = rows[len(rows)-1].ID
+		if len(rows) < batchLimit {
+			return 0, nil
+		}
+	}
+}
+
 func (r *MediaRepository) searchIndexEnabled(ctx context.Context) bool {
 	if r == nil || r.db == nil {
 		return false
 	}
+	if r.db.Dialector == nil || r.db.Dialector.Name() != "sqlite" {
+		return false
+	}
 	r.searchIndexOnce.Do(func() {
 		var count int64
 		err := r.db.WithContext(ctx).

internal/repository/repository_test.go

@@ -1,6 +1,8 @@
 package repository
 
 import (
+	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -70,6 +72,58 @@ func TestMediaUpsertSkipsUnchangedExistingRow(t *testing.T) {
 	}
 }
 
+type fakeMediaSearchBackend struct {
+	ids []string
+	err error
+}
+
+func (f fakeMediaSearchBackend) SearchMediaIDs(context.Context, string, int, int, MediaQueryFilter) ([]string, int64, error) {
+	if f.err != nil {
+		return nil, 0, f.err
+	}
+	return append([]string(nil), f.ids...), int64(len(f.ids)), nil
+}
+
+func TestMediaSearchUsesExternalBackendAndFallsBack(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := database.AutoMigrate(db); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	repos := New(db)
+	lib := model.Library{Name: "Movies", Path: "/media/movie", Type: "movie", Enabled: true}
+	if err := repos.Library.Create(t.Context(), &lib); err != nil {
+		t.Fatal(err)
+	}
+	for _, row := range []model.Media{
+		{Base: model.Base{ID: "m-1"}, LibraryID: lib.ID, Title: "Alpha", Path: "/media/a.mkv"},
+		{Base: model.Base{ID: "m-2"}, LibraryID: lib.ID, Title: "Beta", Path: "/media/b.mkv"},
+	} {
+		if err := repos.DB.Create(&row).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	repos.Media.SetSearchBackend(fakeMediaSearchBackend{ids: []string{"m-2", "m-1"}})
+	items, total, err := repos.Media.SearchFilteredPage(t.Context(), "anything", 0, 10, MediaQueryFilter{IncludeNSFW: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if total != 2 || len(items) != 2 || items[0].ID != "m-2" || items[1].ID != "m-1" {
+		t.Fatalf("external search result total=%d items=%#v", total, items)
+	}
+
+	repos.Media.SetSearchBackend(fakeMediaSearchBackend{err: errors.New("opensearch down")})
+	items, total, err = repos.Media.SearchFilteredPage(t.Context(), "Alpha", 0, 10, MediaQueryFilter{IncludeNSFW: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if total != 1 || len(items) != 1 || items[0].ID != "m-1" {
+		t.Fatalf("fallback result total=%d items=%#v", total, items)
+	}
+}
+
 func TestMediaSearchFilteredSupportsChineseFuzzyTerms(t *testing.T) {
 	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
 	if err != nil {